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

ARTEMIS-1657 - Properly decrement memory usage when moving messages from 

queue

When messages are retried and moved froma DLQ to the original queue the
memory usage tracker needs to be decremented
This commit is contained in:
Christopher L. Shannon (cshannon) 2018-02-02 07:39:50 -05:00 committed by Clebert Suconic
parent 26a28d0686
commit 586487155c
2 changed files with 48 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1813,6 +1813,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
if (!ignored) {
move(null, toAddress, binding, ref, rejectDuplicates, AckReason.NORMAL);
refRemoved(ref);
//move(toAddress, tx, ref, false, rejectDuplicates);
}
}
@ -1861,9 +1862,8 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
move(SimpleString.toSimpleString(originalMessageAddress), tx, ref, false, false, targetQueue.longValue());
} else {
move(SimpleString.toSimpleString(originalMessageAddress), tx, ref, false, false);
}
refRemoved(ref);
}
}
});

46
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/QueueControlTest.java
View File

@ -18,6 +18,7 @@ package org.apache.activemq.artemis.tests.integration.management;
import static org.apache.activemq.artemis.core.management.impl.openmbean.CompositeDataConstants.BODY;
import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
@ -25,6 +26,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import javax.json.JsonArray;
import javax.json.JsonObject;
@ -52,9 +54,11 @@ import org.apache.activemq.artemis.api.core.management.QueueControl;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.DivertConfiguration;
import org.apache.activemq.artemis.core.messagecounter.impl.MessageCounterManagerImpl;
import org.apache.activemq.artemis.core.postoffice.impl.LocalQueueBinding;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.ActiveMQServers;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.impl.QueueImpl;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.tests.integration.jms.server.management.JMSUtil;
import org.apache.activemq.artemis.utils.Base64;
@ -977,6 +981,22 @@ public class QueueControlTest extends ManagementTestBase {
session.start();
final LocalQueueBinding binding = (LocalQueueBinding) server.getPostOffice().getBinding(qName);
Queue q = binding.getQueue();
final LocalQueueBinding binding2 = (LocalQueueBinding) server.getPostOffice().getBinding(dlq);
Queue q2 = binding2.getQueue();
Field queueMemorySizeField = QueueImpl.class.getDeclaredField("queueMemorySize");
queueMemorySizeField.setAccessible(true);
//Get memory size counters to verify
AtomicInteger queueMemorySize1 = (AtomicInteger) queueMemorySizeField.get(q);
AtomicInteger queueMemorySize2 = (AtomicInteger) queueMemorySizeField.get(q2);
//Verify that original queue has a memory size greater than 0 and DLQ is 0
assertTrue(queueMemorySize1.get() > 0);
assertTrue(queueMemorySize2.get() == 0);
// Read and rollback all messages to DLQ
ClientConsumer clientConsumer = session.createConsumer(qName);
for (int i = 0; i < numMessagesToTest; i++) {
@ -989,6 +1009,10 @@ public class QueueControlTest extends ManagementTestBase {
Assert.assertNull(clientConsumer.receiveImmediate());
//Verify that original queue has a memory size of 0 and DLQ is greater than 0 after rollback
assertTrue(queueMemorySize1.get() == 0);
assertTrue(queueMemorySize2.get() > 0);
QueueControl dlqQueueControl = createManagementControl(dla, dlq);
Assert.assertEquals(numMessagesToTest, getMessageCount(dlqQueueControl));
@ -998,6 +1022,10 @@ public class QueueControlTest extends ManagementTestBase {
// Assert DLQ is empty...
Assert.assertEquals(0, getMessageCount(dlqQueueControl));
//Verify that original queue has a memory size of greater than 0 and DLQ is 0 after move
assertTrue(queueMemorySize1.get() > 0);
assertTrue(queueMemorySize2.get() == 0);
// .. and that the messages is now on the original queue once more.
for (int i = 0; i < numMessagesToTest; i++) {
ClientMessage clientMessage = clientConsumer.receive(500);
@ -1007,6 +1035,10 @@ public class QueueControlTest extends ManagementTestBase {
}
clientConsumer.close();
//Verify that original queue and DLQ have a memory size of 0
assertTrue(queueMemorySize1.get() == 0);
assertTrue(queueMemorySize2.get() == 0);
}
/**
@ -1035,14 +1067,28 @@ public class QueueControlTest extends ManagementTestBase {
message.putLongProperty(key, value);
producer.send(message);
final LocalQueueBinding binding = (LocalQueueBinding) server.getPostOffice().getBinding(queue);
Queue q = binding.getQueue();
Field queueMemorySizeField = QueueImpl.class.getDeclaredField("queueMemorySize");
queueMemorySizeField.setAccessible(true);
//Get memory size counters to verify
AtomicInteger queueMemorySize = (AtomicInteger) queueMemorySizeField.get(q);
QueueControl queueControl = createManagementControl(address, queue);
Assert.assertEquals(1, getMessageCount(queueControl));
//verify memory usage is greater than 0
Assert.assertTrue(queueMemorySize.get() > 0);
// moved all messages to otherQueue
int movedMessagesCount = queueControl.moveMessages(null, otherQueue.toString());
Assert.assertEquals(1, movedMessagesCount);
Assert.assertEquals(0, getMessageCount(queueControl));
//verify memory usage is 0 after move
Assert.assertEquals(0, queueMemorySize.get());
// check there is no message to consume from queue
consumeMessages(0, session, queue);