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

AMQ-7129 - fix durable message size statistics with individual ack 

Make sure that the pending message size for a durable sub only includes
messages part of the ack range
This commit is contained in:
Christopher L. Shannon (cshannon) 2019-01-11 09:56:03 -05:00
parent d29c1be66a
commit fa2daa25e9
2 changed files with 27 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
View File

@ -3030,16 +3030,23 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
SequenceSet messageSequences = sd.ackPositions.get(tx, subscriptionKey); SequenceSet messageSequences = sd.ackPositions.get(tx, subscriptionKey);
if (messageSequences != null) { if (messageSequences != null) {
Sequence head = messageSequences.getHead(); if (!messageSequences.isEmpty()) {
if (head != null) { final Sequence head = messageSequences.getHead();
//get an iterator over the order index starting at the first unacked message //get an iterator over the order index starting at the first unacked message
//and go over each message to add up the size //and go over each message to add up the size
Iterator<Entry<Long, MessageKeys>> iterator = sd.orderIndex.iterator(tx, Iterator<Entry<Long, MessageKeys>> iterator = sd.orderIndex.iterator(tx,
new MessageOrderCursor(head.getFirst())); new MessageOrderCursor(head.getFirst()));
final boolean contiguousRange = messageSequences.size() == 1;
while (iterator.hasNext()) { while (iterator.hasNext()) {
Entry<Long, MessageKeys> entry = iterator.next(); Entry<Long, MessageKeys> entry = iterator.next();
locationSize += entry.getValue().location.getSize(); //Verify sequence contains the key
//if contiguous we just add all starting with the first but if not
//we need to check if the id is part of the range - could happen if individual ack mode was used
if (contiguousRange || messageSequences.contains(entry.getKey())) {
locationSize += entry.getValue().location.getSize();
}
} }
} }
} }

17
activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBDurableMessageRecoveryTest.java
View File

@ -209,6 +209,12 @@ public class KahaDBDurableMessageRecoveryTest {
// Verify there are 8 messages left still and restart broker // Verify there are 8 messages left still and restart broker
assertTrue(Wait.waitFor(() -> 8 == getPendingMessageCount(topic, "clientId1", "sub1"), 3000, 500)); assertTrue(Wait.waitFor(() -> 8 == getPendingMessageCount(topic, "clientId1", "sub1"), 3000, 500));
assertTrue(Wait.waitFor(() -> 10 == getPendingMessageCount(topic, "clientId1", "sub2"), 3000, 500)); assertTrue(Wait.waitFor(() -> 10 == getPendingMessageCount(topic, "clientId1", "sub2"), 3000, 500));
//Verify the pending size is less for sub1
assertTrue(getPendingMessageSize(topic, "clientId1", "sub1") > 0);
assertTrue(getPendingMessageSize(topic, "clientId1", "sub2") > 0);
assertTrue(getPendingMessageSize(topic, "clientId1", "sub1") < getPendingMessageSize(topic, "clientId1", "sub2"));
subscriber1.close(); subscriber1.close();
subscriber2.close(); subscriber2.close();
restartBroker(recoverIndex); restartBroker(recoverIndex);
@ -217,6 +223,11 @@ public class KahaDBDurableMessageRecoveryTest {
assertTrue(Wait.waitFor(() -> 8 == getPendingMessageCount(topic, "clientId1", "sub1"), 3000, 500)); assertTrue(Wait.waitFor(() -> 8 == getPendingMessageCount(topic, "clientId1", "sub1"), 3000, 500));
assertTrue(Wait.waitFor(() -> 10 == getPendingMessageCount(topic, "clientId1", "sub2"), 3000, 500)); assertTrue(Wait.waitFor(() -> 10 == getPendingMessageCount(topic, "clientId1", "sub2"), 3000, 500));
//Verify the pending size is less for sub1
assertTrue(getPendingMessageSize(topic, "clientId1", "sub1") > 0);
assertTrue(getPendingMessageSize(topic, "clientId1", "sub2") > 0);
assertTrue(getPendingMessageSize(topic, "clientId1", "sub1") < getPendingMessageSize(topic, "clientId1", "sub2"));
// Recreate subscriber and try and receive the other 8 messages // Recreate subscriber and try and receive the other 8 messages
session = getSession(ActiveMQSession.AUTO_ACKNOWLEDGE); session = getSession(ActiveMQSession.AUTO_ACKNOWLEDGE);
subscriber1 = session.createDurableSubscriber(topic, "sub1"); subscriber1 = session.createDurableSubscriber(topic, "sub1");
@ -347,4 +358,10 @@ public class KahaDBDurableMessageRecoveryTest {
final TopicMessageStore store = (TopicMessageStore) brokerTopic.getMessageStore(); final TopicMessageStore store = (TopicMessageStore) brokerTopic.getMessageStore();
return store.getMessageCount(clientId, subId); return store.getMessageCount(clientId, subId);
} }
protected long getPendingMessageSize(ActiveMQTopic topic, String clientId, String subId) throws Exception {
final Topic brokerTopic = (Topic) broker.getDestination(topic);
final TopicMessageStore store = (TopicMessageStore) brokerTopic.getMessageStore();
return store.getMessageSize(clientId, subId);
}
} }