Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[AMQ-8068] Fix topic memory leak on message eviction using UniquePropertyMessageEvictionStrategy 

(cherry picked from commit 2bc87c2c5f)
This commit is contained in:
jbonofre 2020-11-22 11:49:34 +01:00
parent acae93b93e
commit d9b2d9309e
2 changed files with 15 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/UniquePropertyMessageEvictionStrategy.java
View File

@ -22,7 +22,7 @@ import java.io.IOException;
import java.util.HashMap; import java.util.HashMap;
import java.util.Iterator; import java.util.Iterator;
import java.util.LinkedList; import java.util.LinkedList;
import java.util.List;
/** /**
* An eviction strategy which evicts the oldest message within messages with the same property value * An eviction strategy which evicts the oldest message within messages with the same property value
@ -45,12 +45,10 @@ public class UniquePropertyMessageEvictionStrategy extends MessageEvictionStrate
@Override @Override
public MessageReference[] evictMessages(LinkedList messages) throws IOException { public MessageReference[] evictMessages(LinkedList messages) throws IOException {
MessageReference oldest = (MessageReference)messages.getFirst(); List<MessageReference> messageReferences = new LinkedList<>(messages);
HashMap<Object, MessageReference> pivots = new HashMap<Object, MessageReference>(); HashMap<Object, MessageReference> pivots = new HashMap<>();
Iterator iter = messages.iterator();
for (int i = 0; iter.hasNext(); i++) { for (MessageReference reference : messageReferences) {
MessageReference reference = (MessageReference) iter.next();
if (propertyName != null && reference.getMessage().getProperty(propertyName) != null) { if (propertyName != null && reference.getMessage().getProperty(propertyName) != null) {
Object key = reference.getMessage().getProperty(propertyName); Object key = reference.getMessage().getProperty(propertyName);
if (pivots.containsKey(key)) { if (pivots.containsKey(key)) {
@ -64,16 +62,13 @@ public class UniquePropertyMessageEvictionStrategy extends MessageEvictionStrate
} }
} }
if (!pivots.isEmpty()) { if (pivots.isEmpty() || pivots.values().size() == messageReferences.size()) {
for (MessageReference ref : pivots.values()) { return new MessageReference[]{(MessageReference) messages.removeFirst()};
messages.remove(ref); } else {
messages.removeIf(message -> !pivots.containsValue(message));
messageReferences.removeAll(pivots.values());
return messageReferences.toArray(new MessageReference[0]);
}
} }
if (messages.size() != 0) {
return (MessageReference[])messages.toArray(new MessageReference[messages.size()]);
}
}
return new MessageReference[] {oldest};
}
} }

6
activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/UniquePropertyMessageEvictionStrategyTest.java
View File

@ -18,11 +18,13 @@
package org.apache.activemq.broker.region; package org.apache.activemq.broker.region;
import org.apache.activemq.EmbeddedBrokerTestSupport; import org.apache.activemq.EmbeddedBrokerTestSupport;
import org.apache.activemq.TestSupport;
import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.region.policy.ConstantPendingMessageLimitStrategy; import org.apache.activemq.broker.region.policy.ConstantPendingMessageLimitStrategy;
import org.apache.activemq.broker.region.policy.PolicyEntry; import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap; import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.broker.region.policy.UniquePropertyMessageEvictionStrategy; import org.apache.activemq.broker.region.policy.UniquePropertyMessageEvictionStrategy;
import org.apache.activemq.command.ActiveMQDestination;
import javax.jms.*; import javax.jms.*;
import java.util.ArrayList; import java.util.ArrayList;
@ -30,8 +32,6 @@ import java.util.List;
public class UniquePropertyMessageEvictionStrategyTest extends EmbeddedBrokerTestSupport { public class UniquePropertyMessageEvictionStrategyTest extends EmbeddedBrokerTestSupport {
@Override @Override
protected BrokerService createBroker() throws Exception { protected BrokerService createBroker() throws Exception {
BrokerService broker = super.createBroker(); BrokerService broker = super.createBroker();
@ -81,7 +81,6 @@ public class UniquePropertyMessageEvictionStrategyTest extends EmbeddedBrokerTes
} }
} }
for (int i = 0; i < 11; i++) { for (int i = 0; i < 11; i++) {
javax.jms.Message msg = consumer.receive(1000); javax.jms.Message msg = consumer.receive(1000);
assertNotNull(msg); assertNotNull(msg);
@ -100,6 +99,7 @@ public class UniquePropertyMessageEvictionStrategyTest extends EmbeddedBrokerTes
javax.jms.Message msg = consumer.receive(1000); javax.jms.Message msg = consumer.receive(1000);
assertNull(msg); assertNull(msg);
assertEquals("usage goes to 0", 0, TestSupport.getDestination(broker, ActiveMQDestination.transform(destination)).getMemoryUsage().getUsage());
} }
} }