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

Merge pull request #583 from jbonofre/AMQ-8068 

[AMQ-8068] Fix topic memory leak on message eviction using UniquePropertyMessageEvictionStrategy
This commit is contained in:
Jean-Baptiste Onofré 2020-11-23 13:40:43 +01:00 committed by GitHub
parent 1a6aede5a1 2bc87c2c5f
commit 4aa298811e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
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.Iterator;
import java.util.LinkedList;
import java.util.List;
/**
* 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
public MessageReference[] evictMessages(LinkedList messages) throws IOException {
MessageReference oldest = (MessageReference)messages.getFirst();
HashMap<Object, MessageReference> pivots = new HashMap<Object, MessageReference>();
Iterator iter = messages.iterator();
List<MessageReference> messageReferences = new LinkedList<>(messages);
HashMap<Object, MessageReference> pivots = new HashMap<>();
for (int i = 0; iter.hasNext(); i++) {
MessageReference reference = (MessageReference) iter.next();
for (MessageReference reference : messageReferences) {
if (propertyName != null && reference.getMessage().getProperty(propertyName) != null) {
Object key = reference.getMessage().getProperty(propertyName);
if (pivots.containsKey(key)) {
@ -64,16 +62,13 @@ public class UniquePropertyMessageEvictionStrategy extends MessageEvictionStrate
}
}
if (!pivots.isEmpty()) {
for (MessageReference ref : pivots.values()) {
messages.remove(ref);
}
if (messages.size() != 0) {
return (MessageReference[])messages.toArray(new MessageReference[messages.size()]);
}
if (pivots.isEmpty() || pivots.values().size() == messageReferences.size()) {
return new MessageReference[]{(MessageReference) messages.removeFirst()};
} else {
messages.removeIf(message -> !pivots.containsValue(message));
messageReferences.removeAll(pivots.values());
return messageReferences.toArray(new MessageReference[0]);
}
return new MessageReference[] {oldest};
}
}

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