This is an automated email from the ASF dual-hosted git repository.
cshannon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/master by this push:
new fa2daa2 AMQ-7129 - fix durable message size statistics with
individual ack
fa2daa2 is described below
commit fa2daa25e9acd3f37bb1ee0d37717d2383e67a85
Author: Christopher L. Shannon (cshannon) <christopher.l.shan...@gmail.com>
AuthorDate: Fri Jan 11 09:56:03 2019 -0500
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
---
.../apache/activemq/store/kahadb/MessageDatabase.java | 13 ++++++++++---
.../store/kahadb/KahaDBDurableMessageRecoveryTest.java | 17 +++++++++++++++++
2 files changed, 27 insertions(+), 3 deletions(-)
diff --git
a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
index 5d9b548..0bc6e81 100644
---
a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
+++
b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
@@ -3030,16 +3030,23 @@ public abstract class MessageDatabase extends
ServiceSupport implements BrokerSe
SequenceSet messageSequences = sd.ackPositions.get(tx,
subscriptionKey);
if (messageSequences != null) {
- Sequence head = messageSequences.getHead();
- if (head != null) {
+ if (!messageSequences.isEmpty()) {
+ final Sequence head = messageSequences.getHead();
+
//get an iterator over the order index starting at the
first unacked message
//and go over each message to add up the size
Iterator<Entry<Long, MessageKeys>> iterator =
sd.orderIndex.iterator(tx,
new MessageOrderCursor(head.getFirst()));
+ final boolean contiguousRange = messageSequences.size() ==
1;
while (iterator.hasNext()) {
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();
+ }
}
}
}
diff --git
a/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBDurableMessageRecoveryTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBDurableMessageRecoveryTest.java
index a44e8c0..519648e 100644
---
a/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBDurableMessageRecoveryTest.java
+++
b/activemq-unit-tests/src/test/java/org/apache/activemq/store/kahadb/KahaDBDurableMessageRecoveryTest.java
@@ -209,6 +209,12 @@ public class KahaDBDurableMessageRecoveryTest {
// Verify there are 8 messages left still and restart broker
assertTrue(Wait.waitFor(() -> 8 == getPendingMessageCount(topic,
"clientId1", "sub1"), 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();
subscriber2.close();
restartBroker(recoverIndex);
@@ -217,6 +223,11 @@ public class KahaDBDurableMessageRecoveryTest {
assertTrue(Wait.waitFor(() -> 8 == getPendingMessageCount(topic,
"clientId1", "sub1"), 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
session = getSession(ActiveMQSession.AUTO_ACKNOWLEDGE);
subscriber1 = session.createDurableSubscriber(topic, "sub1");
@@ -347,4 +358,10 @@ public class KahaDBDurableMessageRecoveryTest {
final TopicMessageStore store = (TopicMessageStore)
brokerTopic.getMessageStore();
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);
+ }
}
