activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gtu...@apache.org
Subject activemq git commit: [AMQ-6562] - suppress warn of durable sub duplicate from the store on cache exhaustion - expected in the absense of ordered sequenceid and setBatch. Fix leak of duplicates pending processing on batch fill for the durable sub case and
Date Fri, 13 Jan 2017 11:07:48 GMT
Repository: activemq
Updated Branches:
  refs/heads/master d9c74d731 -> 511b9b642


[AMQ-6562] - suppress warn of durable sub duplicate from the store on cache exhaustion - expected
in the absense of ordered sequenceid and setBatch. Fix leak of duplicates pending processing
on batch fill for the durable sub case and remove eager page in for prefetch=0


Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/511b9b64
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/511b9b64
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/511b9b64

Branch: refs/heads/master
Commit: 511b9b642a198fe98409e1f1aac6176b9420d951
Parents: d9c74d7
Author: gtully <gary.tully@gmail.com>
Authored: Fri Jan 13 11:04:48 2017 +0000
Committer: gtully <gary.tully@gmail.com>
Committed: Fri Jan 13 11:06:05 2017 +0000

----------------------------------------------------------------------
 .../broker/region/PrefetchSubscription.java     |   2 +-
 .../region/cursors/AbstractStoreCursor.java     |   9 +-
 .../region/cursors/TopicStorePrefetch.java      |   9 +
 .../activemq/usecases/DurableSubCacheTest.java  | 197 +++++++++++++++++++
 4 files changed, 214 insertions(+), 3 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq/blob/511b9b64/activemq-broker/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
index 0a277fb..db133c1 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
@@ -684,7 +684,7 @@ public abstract class PrefetchSubscription extends AbstractSubscription
{
                     setPendingBatchSize(pending, numberToDispatch);
                     int count = 0;
                     pending.reset();
-                    while (pending.hasNext() && !isFull() && count < numberToDispatch)
{
+                    while (count < numberToDispatch && !isFull() && pending.hasNext())
{
                         MessageReference node = pending.next();
                         if (node == null) {
                             break;

http://git-wip-us.apache.org/repos/asf/activemq/blob/511b9b64/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractStoreCursor.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractStoreCursor.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractStoreCursor.java
index 4d7ffea..0295b33 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractStoreCursor.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractStoreCursor.java
@@ -110,8 +110,7 @@ public abstract class AbstractStoreCursor extends AbstractPendingMessageCursor
i
             recovered = true;
         } else if (!cached) {
             // a duplicate from the store (!cached) - needs to be removed/acked - otherwise
it will get re dispatched on restart
-            if (message.isRecievedByDFBridge()) {
-                // expected for messages pending acks with kahadb.concurrentStoreAndDispatchQueues=true
+            if (duplicateFromStoreExcepted(message)) {
                 if (LOG.isTraceEnabled()) {
                     LOG.trace("{} store replayed pending message due to concurrentStoreAndDispatchQueues
{} seq: {}", this, message.getMessageId(), message.getMessageId().getFutureOrSequenceLong());
                 }
@@ -128,6 +127,12 @@ public abstract class AbstractStoreCursor extends AbstractPendingMessageCursor
i
         return recovered;
     }
 
+    protected boolean duplicateFromStoreExcepted(Message message) {
+        // expected for messages pending acks with kahadb.concurrentStoreAndDispatchQueues=true
for
+        // which this existing unused flag has been repurposed
+        return message.isRecievedByDFBridge();
+    }
+
     public static boolean gotToTheStore(Message message) throws Exception {
         if (message.isRecievedByDFBridge()) {
             // concurrent store and dispatch - wait to see if the message gets to the store
to see

http://git-wip-us.apache.org/repos/asf/activemq/blob/511b9b64/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/TopicStorePrefetch.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/TopicStorePrefetch.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/TopicStorePrefetch.java
index bfc745e..71bb4eb 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/TopicStorePrefetch.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/TopicStorePrefetch.java
@@ -95,6 +95,14 @@ class TopicStorePrefetch extends AbstractStoreCursor {
     }
 
     @Override
+    protected boolean duplicateFromStoreExcepted(Message message) {
+        // setBatch is not implemented - sequence order not reliable with concurrent transactions
+        // on cache exhaustion - first pageIn starts from last ack location which may replay
what
+        // cursor has dispatched
+        return true;
+    }
+
+    @Override
     protected synchronized int getStoreSize() {
         try {
             return store.getMessageCount(clientId, subscriberName);
@@ -137,6 +145,7 @@ class TopicStorePrefetch extends AbstractStoreCursor {
         this.storeHasMessages = false;
         this.store.recoverNextMessages(clientId, subscriberName,
                 maxBatchSize, this);
+        dealWithDuplicates();
         if (!this.storeHasMessages && (!this.batchList.isEmpty() || !hadSpace)) {
             this.storeHasMessages = true;
         }

http://git-wip-us.apache.org/repos/asf/activemq/blob/511b9b64/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubCacheTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubCacheTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubCacheTest.java
new file mode 100644
index 0000000..17d1931
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/DurableSubCacheTest.java
@@ -0,0 +1,197 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.usecases;
+
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.cursors.AbstractStoreCursor;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.activemq.util.DefaultTestAppender;
+import org.apache.log4j.Appender;
+import org.apache.log4j.Level;
+import org.apache.log4j.spi.LoggingEvent;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.Connection;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.TopicSubscriber;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.fail;
+
+public class DurableSubCacheTest {
+    private static final Logger LOG = LoggerFactory.getLogger(DurableSubCacheTest.class);
+
+
+    private final ActiveMQTopic topic = new ActiveMQTopic("T1");
+    private BrokerService broker;
+
+    @Before
+    public void setUp() throws Exception {
+
+        broker = createAndStartBroker();
+        broker.waitUntilStarted();
+    }
+
+
+    private BrokerService createAndStartBroker()
+            throws Exception {
+        BrokerService broker = new BrokerService();
+        broker.setDeleteAllMessagesOnStartup(true);
+        broker.setUseJmx(false);
+        broker.setAdvisorySupport(false);
+        broker.getSystemUsage().getMemoryUsage().setLimit(100 * 1024);
+
+        PolicyMap policyMap = new PolicyMap();
+        PolicyEntry policy = new PolicyEntry();
+        policy.setCursorMemoryHighWaterMark(20);
+        policyMap.put(topic, policy);
+        broker.setDestinationPolicy(policyMap);
+
+        broker.start();
+
+        return broker;
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        broker.stop();
+    }
+
+    @Test
+    public void testCacheExhaustion() throws Exception {
+        doTestCacheExhaustion(1000);
+    }
+
+    @Test
+    public void testCacheExhaustionPrefetch0() throws Exception {
+        doTestCacheExhaustion(0);
+    }
+
+    public void doTestCacheExhaustion(int prefetch) throws Exception {
+
+        createDurableSub(topic, "my_sub_1");
+
+        publishMesssages(topic, 20);
+
+        org.apache.log4j.Logger log4jLogger = org.apache.log4j.Logger.getLogger(AbstractStoreCursor.class.getCanonicalName());
+        final AtomicBoolean failed = new AtomicBoolean(false);
+
+        Appender appender = new DefaultTestAppender() {
+            @Override
+            public void doAppend(LoggingEvent event) {
+                if (event.getLevel() == Level.WARN) {
+                    LOG.info("Got warn event:" + event.getRenderedMessage());
+                    failed.set(true);
+                }
+            }
+        };
+        log4jLogger.addAppender(appender);
+
+        try {
+            consumeDurableSub(topic, "my_sub_1", 20, prefetch);
+        } finally {
+            log4jLogger.removeAppender(appender);
+        }
+
+        assertFalse("no warning from the cursor", failed.get());
+    }
+
+    private void publishMesssages(ActiveMQTopic topic, int messageCount) throws Exception
{
+
+        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(broker.getVmConnectorURI());
+        connectionFactory.setWatchTopicAdvisories(false);
+        Connection con = connectionFactory.createConnection();
+        con.start();
+
+        Session session = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        MessageProducer producer = session.createProducer(topic);
+
+        try {
+            String textMessage = new String(new byte[1024]);
+            TextMessage msg = session.createTextMessage(textMessage);
+
+            for (int i = 0; i < messageCount; i++) {
+                producer.send(msg);
+            }
+        } finally {
+            con.close();
+        }
+
+    }
+
+
+    private void createDurableSub(ActiveMQTopic topic, String subID) throws Exception {
+
+
+        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(broker.getVmConnectorURI());
+        connectionFactory.setWatchTopicAdvisories(false);
+        Connection con = connectionFactory.createConnection();
+        con.setClientID("CONNECTION-" + subID);
+        con.start();
+
+        Session session = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+        session.createDurableSubscriber(topic, subID, null, true);
+        session.close();
+        con.close();
+    }
+
+    private void consumeDurableSub(ActiveMQTopic topic, String subID, int messageCount) throws
Exception {
+        consumeDurableSub(topic, subID, messageCount, 1000);
+    }
+
+    private void consumeDurableSub(ActiveMQTopic topic, String subID, int messageCount, int
prefetch) throws Exception {
+
+        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(broker.getVmConnectorURI());
+        ActiveMQConnection con = (ActiveMQConnection) connectionFactory.createConnection();
+        con.setClientID("CONNECTION-" + subID);
+        con.getPrefetchPolicy().setAll(prefetch);
+        con.start();
+
+        Session session = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+        TopicSubscriber topicSubscriber
+                = session.createDurableSubscriber(topic, subID, null, true);
+
+        try {
+
+            for (int i = 0; i < messageCount; i++) {
+                javax.jms.Message message = topicSubscriber.receive(4000l);
+                if (message == null) {
+                    fail("should have received a message");
+                }
+            }
+
+        } finally {
+            con.close();
+        }
+    }
+
+
+}
\ No newline at end of file


Mime
View raw message