You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cs...@apache.org on 2016/02/04 16:18:03 UTC
activemq git commit: https://issues.apache.org/jira/browse/AMQ-6158
Repository: activemq
Updated Branches:
refs/heads/master c17b7fdc7 -> 25ff5699f
https://issues.apache.org/jira/browse/AMQ-6158
Computing messageSize for a durable subscription in KahaDB now runs much
faster (n vs n^2) which is noticable when there are a large number of
pending messages for a durable subscription.
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/25ff5699
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/25ff5699
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/25ff5699
Branch: refs/heads/master
Commit: 25ff5699f1fb7c73668d5da2eacb53ad2ef14289
Parents: c17b7fd
Author: Christopher L. Shannon (cshannon) <ch...@gmail.com>
Authored: Thu Feb 4 15:16:11 2016 +0000
Committer: Christopher L. Shannon (cshannon) <ch...@gmail.com>
Committed: Thu Feb 4 15:17:49 2016 +0000
----------------------------------------------------------------------
.../activemq/store/kahadb/MessageDatabase.java | 25 ++++++--------
.../AbstractPendingMessageCursorTest.java | 6 ++++
.../cursors/KahaDBPendingMessageCursorTest.java | 36 ++++++++++++++++++++
3 files changed, 53 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/25ff5699/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
----------------------------------------------------------------------
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 c7e1a40..931a18b 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
@@ -2588,31 +2588,28 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe
}
public long getStoredMessageSize(Transaction tx, StoredDestination sd, String subscriptionKey) throws IOException {
+ //grab the messages attached to this subscription
SequenceSet messageSequences = sd.ackPositions.get(tx, subscriptionKey);
+
long locationSize = 0;
if (messageSequences != null) {
- Iterator<Long> sequences = messageSequences.iterator();
+ Sequence head = messageSequences.getHead();
+ if (head != null) {
+ //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()));
- while (sequences.hasNext()) {
- Long sequenceId = sequences.next();
- //the last item is the next marker
- if (!sequences.hasNext()) {
- break;
- }
- Iterator<Entry<Location, Long>> iterator = sd.locationIndex.iterator(tx);
while (iterator.hasNext()) {
- Entry<Location, Long> entry = iterator.next();
- if (entry.getValue() == sequenceId - 1) {
- locationSize += entry.getKey().getSize();
- break;
- }
-
+ Entry<Long, MessageKeys> entry = iterator.next();
+ locationSize += entry.getValue().location.getSize();
}
}
}
return locationSize;
}
+
protected String key(KahaDestination destination) {
return destination.getType().getNumber() + ":" + destination.getName();
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/25ff5699/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/AbstractPendingMessageCursorTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/AbstractPendingMessageCursorTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/AbstractPendingMessageCursorTest.java
index f496806..a31f402 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/AbstractPendingMessageCursorTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/AbstractPendingMessageCursorTest.java
@@ -16,6 +16,8 @@
*/
package org.apache.activemq.broker.region.cursors;
+import static org.junit.Assert.assertEquals;
+
import java.io.IOException;
import java.net.URI;
import java.util.concurrent.atomic.AtomicLong;
@@ -276,6 +278,10 @@ public abstract class AbstractPendingMessageCursorTest extends AbstractStoreStat
verifyPendingStats(dest, subKey, 200, publishedMessageSize.get());
verifyStoreStats(dest, 200, publishedMessageSize.get());
+ //should be equal in this case
+ assertEquals(dest.getDurableTopicSubs().get(subKey).getPendingMessageSize(),
+ dest.getMessageStore().getMessageStoreStatistics().getMessageSize().getTotalSize());
+
//consume all messages
consumeDurableTestMessages(connection, "sub1", 200, publishedMessageSize);
http://git-wip-us.apache.org/repos/asf/activemq/blob/25ff5699/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/KahaDBPendingMessageCursorTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/KahaDBPendingMessageCursorTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/KahaDBPendingMessageCursorTest.java
index e0921dc..8a2c287 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/KahaDBPendingMessageCursorTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/region/cursors/KahaDBPendingMessageCursorTest.java
@@ -16,6 +16,8 @@
*/
package org.apache.activemq.broker.region.cursors;
+import static org.junit.Assert.assertEquals;
+
import java.io.File;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicLong;
@@ -84,6 +86,10 @@ public class KahaDBPendingMessageCursorTest extends
verifyPendingStats(topic, subKey, 200, publishedMessageSize.get());
verifyStoreStats(topic, 200, publishedMessageSize.get());
+ //should be equal in this case
+ assertEquals(topic.getDurableTopicSubs().get(subKey).getPendingMessageSize(),
+ topic.getMessageStore().getMessageStoreStatistics().getMessageSize().getTotalSize());
+
// stop, restart broker and publish more messages
stopBroker();
this.setUpBroker(false);
@@ -101,6 +107,36 @@ public class KahaDBPendingMessageCursorTest extends
}
+ @Test(timeout=60000)
+ public void testMessageSizeTwoDurablesPartialConsumption() throws Exception {
+ AtomicLong publishedMessageSize = new AtomicLong();
+
+ Connection connection = new ActiveMQConnectionFactory(brokerConnectURI).createConnection();
+ connection.setClientID("clientId");
+ connection.start();
+
+ SubscriptionKey subKey = new SubscriptionKey("clientId", "sub1");
+ SubscriptionKey subKey2 = new SubscriptionKey("clientId", "sub2");
+ org.apache.activemq.broker.region.Topic dest = publishTestMessagesDurable(
+ connection, new String[] {"sub1", "sub2"}, 200, publishedMessageSize, DeliveryMode.PERSISTENT);
+
+ //verify the count and size - durable is offline so all 200 should be pending since none are in prefetch
+ verifyPendingStats(dest, subKey, 200, publishedMessageSize.get());
+ verifyStoreStats(dest, 200, publishedMessageSize.get());
+
+ //consume all messages
+ consumeDurableTestMessages(connection, "sub1", 50, publishedMessageSize);
+
+ //150 should be left
+ verifyPendingStats(dest, subKey, 150, publishedMessageSize.get());
+
+ //200 should be left
+ verifyPendingStats(dest, subKey2, 200, publishedMessageSize.get());
+ verifyStoreStats(dest, 200, publishedMessageSize.get());
+
+ connection.close();
+ }
+
/**
* Test that the the counter restores size and works after restart and more
* messages are published