ARTEMIS-1800 - Fix metrics decrement on scheduled message cancel

The queue metrics were being decremented improperly because on iteration
over the cancelled scheduled messages because the flag for fromMessageReferences was not
set to false. Setting the flag to false skips over the metrics update
which is what we want as the scheduled messages were never added to the
message references in the first place so the metrics don't need updating
This commit is contained in:
Christopher L. Shannon (cshannon) 2018-04-10 15:54:29 -04:00
parent fa3e37fdb9
commit 70f0908b4e
2 changed files with 34 additions and 1 deletions

View File

@ -1546,7 +1546,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
List<MessageReference> cancelled = scheduledDeliveryHandler.cancel(filter1);
for (MessageReference messageReference : cancelled) {
messageAction.actMessage(tx, messageReference);
messageAction.actMessage(tx, messageReference, false);
count++;
txCount++;
}

View File

@ -2510,6 +2510,39 @@ public class QueueControlTest extends ManagementTestBase {
Assert.assertEquals(new String(body), "theBody");
}
@Test
public void testGetScheduledCountOnRemove() throws Exception {
long delay = Integer.MAX_VALUE;
SimpleString address = RandomUtil.randomSimpleString();
SimpleString queue = RandomUtil.randomSimpleString();
session.createQueue(address, RoutingType.MULTICAST, queue, null, durable);
QueueControl queueControl = createManagementControl(address, queue);
Assert.assertEquals(0, queueControl.getScheduledCount());
Field queueMemorySizeField = QueueImpl.class.getDeclaredField("queueMemorySize");
queueMemorySizeField.setAccessible(true);
final LocalQueueBinding binding = (LocalQueueBinding) server.getPostOffice().getBinding(queue);
Queue q = binding.getQueue();
AtomicInteger queueMemorySize1 = (AtomicInteger) queueMemorySizeField.get(q);
assertTrue(queueMemorySize1.get() == 0);
ClientProducer producer = session.createProducer(address);
ClientMessage message = session.createMessage(durable);
message.putLongProperty(Message.HDR_SCHEDULED_DELIVERY_TIME, System.currentTimeMillis() + delay);
producer.send(message);
queueControl.removeAllMessages();
Assert.assertEquals(0, queueControl.getMessageCount());
//Verify that original queue has a memory size of 0
assertTrue(queueMemorySize1.get() == 0);
session.deleteQueue(queue);
}
// Package protected ---------------------------------------------
// Protected -----------------------------------------------------