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

ARTEMIS-2478 Expired message not removed in non destructive queue

This commit is contained in:
Wei Yang 2019-09-09 18:45:48 +08:00 committed by Clebert Suconic
parent a8d68d9dd7
commit e43c5390cf
2 changed files with 11 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
View File

@ -2745,9 +2745,6 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
logger.trace("Reference " + ref + " being expired");
}
removeMessageReference(holder, ref);
handled++;
consumers.reset();
continue;
@ -2778,8 +2775,9 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
deliveriesInTransit.countUp();
removeMessageReference(holder, ref);
if (!nonDestructive) {
removeMessageReference(holder, ref);
}
ref.setInDelivery(true);
handledconsumer = consumer;
handled++;
@ -2836,10 +2834,8 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
}
protected void removeMessageReference(ConsumerHolder<? extends Consumer> holder, MessageReference ref) {
if (!nonDestructive) {
holder.iter.remove();
refRemoved(ref);
}
holder.iter.remove();
refRemoved(ref);
}
private void checkDepage() {

10
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/JMSNonDestructiveTest.java
View File

@ -53,14 +53,16 @@ public class JMSNonDestructiveTest extends JMSClientTestSupport {
private ConnectionSupplier CoreConnection = () -> createCoreConnection();
protected final boolean persistenceEnabled;
protected final long scanPeriod;
public JMSNonDestructiveTest(boolean persistenceEnabled) {
public JMSNonDestructiveTest(boolean persistenceEnabled, long scanPeriod) {
this.persistenceEnabled = persistenceEnabled;
this.scanPeriod = scanPeriod;
}
@Parameterized.Parameters(name = "persistenceEnabled={0}")
@Parameterized.Parameters(name = "persistenceEnabled={0}, scanPeriod={1}")
public static Collection<Object[]> data() {
Object[][] params = new Object[][]{{false}, {true}};
Object[][] params = new Object[][]{{false, 100}, {true, 100}, {true, -1}};
return Arrays.asList(params);
}
@ -72,7 +74,7 @@ public class JMSNonDestructiveTest extends JMSClientTestSupport {
@Override
protected void addConfiguration(ActiveMQServer server) {
server.getConfiguration().setPersistenceEnabled(persistenceEnabled);
server.getConfiguration().setMessageExpiryScanPeriod(100);
server.getConfiguration().setMessageExpiryScanPeriod(scanPeriod);
server.getAddressSettingsRepository().addMatch(NON_DESTRUCTIVE_QUEUE_NAME, new AddressSettings().setDefaultNonDestructive(true));
server.getAddressSettingsRepository().addMatch(NON_DESTRUCTIVE_EXPIRY_QUEUE_NAME, new AddressSettings().setDefaultNonDestructive(true).setExpiryDelay(100L));
server.getAddressSettingsRepository().addMatch(NON_DESTRUCTIVE_LVQ_QUEUE_NAME, new AddressSettings().setDefaultLastValueQueue(true).setDefaultNonDestructive(true));