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-6854] log warn if page in is blocked on usage preventing dispatch. Reuse blockedProducerWarningInterval to tweak reporting
Date Thu, 02 Nov 2017 16:04:47 GMT
Repository: activemq
Updated Branches:
  refs/heads/master 03b19b9da -> b29ccf348


[AMQ-6854] log warn if page in is blocked on usage preventing dispatch. Reuse blockedProducerWarningInterval
to tweak reporting


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

Branch: refs/heads/master
Commit: b29ccf348803249b9d62cc607a8c7e22f06cf8a5
Parents: 03b19b9
Author: gtully <gary.tully@gmail.com>
Authored: Thu Nov 2 16:04:30 2017 +0000
Committer: gtully <gary.tully@gmail.com>
Committed: Thu Nov 2 16:04:30 2017 +0000

----------------------------------------------------------------------
 .../apache/activemq/broker/region/Queue.java    |  59 +++----
 .../usecases/UsageBlockedDispatchTest.java      | 156 +++++++++++++++++++
 2 files changed, 188 insertions(+), 27 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq/blob/b29ccf34/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
index 04ef3fd..c43a8ba 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
@@ -2029,38 +2029,43 @@ public class Queue extends BaseDestination implements Task, UsageListener,
Index
             } finally {
                 messagesLock.writeLock().unlock();
             }
-            // Only add new messages, not already pagedIn to avoid multiple
-            // dispatch attempts
-            pagedInMessagesLock.writeLock().lock();
-            try {
-                if(isPrioritizedMessages()) {
-                    resultList = new PrioritizedPendingList();
-                } else {
-                    resultList = new OrderedPendingList();
-                }
-                for (QueueMessageReference ref : result) {
-                    if (!pagedInMessages.contains(ref)) {
-                        pagedInMessages.addMessageLast(ref);
-                        resultList.addMessageLast(ref);
+
+            if (count > 0) {
+                // Only add new messages, not already pagedIn to avoid multiple
+                // dispatch attempts
+                pagedInMessagesLock.writeLock().lock();
+                try {
+                    if (isPrioritizedMessages()) {
+                        resultList = new PrioritizedPendingList();
                     } else {
-                        ref.decrementReferenceCount();
-                        // store should have trapped duplicate in it's index, or cursor audit
trapped insert
-                        // or producerBrokerExchange suppressed send.
-                        // note: jdbc store will not trap unacked messages as a duplicate
b/c it gives each message a unique sequence id
-                        LOG.warn("{}, duplicate message {} - {} from cursor, is cursor audit
disabled or too constrained? Redirecting to dlq", this, ref.getMessageId(), ref.getMessage().getMessageId().getFutureOrSequenceLong());
-                        if (store != null) {
-                            ConnectionContext connectionContext = createConnectionContext();
-                            dropMessage(ref);
-                            if (gotToTheStore(ref.getMessage())) {
-                                LOG.debug("Duplicate message {} from cursor, removing from
store", this, ref.getMessage());
-                                store.removeMessage(connectionContext, new MessageAck(ref.getMessage(),
MessageAck.POSION_ACK_TYPE, 1));
+                        resultList = new OrderedPendingList();
+                    }
+                    for (QueueMessageReference ref : result) {
+                        if (!pagedInMessages.contains(ref)) {
+                            pagedInMessages.addMessageLast(ref);
+                            resultList.addMessageLast(ref);
+                        } else {
+                            ref.decrementReferenceCount();
+                            // store should have trapped duplicate in it's index, or cursor
audit trapped insert
+                            // or producerBrokerExchange suppressed send.
+                            // note: jdbc store will not trap unacked messages as a duplicate
b/c it gives each message a unique sequence id
+                            LOG.warn("{}, duplicate message {} - {} from cursor, is cursor
audit disabled or too constrained? Redirecting to dlq", this, ref.getMessageId(), ref.getMessage().getMessageId().getFutureOrSequenceLong());
+                            if (store != null) {
+                                ConnectionContext connectionContext = createConnectionContext();
+                                dropMessage(ref);
+                                if (gotToTheStore(ref.getMessage())) {
+                                    LOG.debug("Duplicate message {} from cursor, removing
from store", this, ref.getMessage());
+                                    store.removeMessage(connectionContext, new MessageAck(ref.getMessage(),
MessageAck.POSION_ACK_TYPE, 1));
+                                }
+                                broker.getRoot().sendToDeadLetterQueue(connectionContext,
ref.getMessage(), null, new Throwable("duplicate paged in from cursor for " + destination));
                             }
-                            broker.getRoot().sendToDeadLetterQueue(connectionContext, ref.getMessage(),
null, new Throwable("duplicate paged in from cursor for " + destination));
                         }
                     }
+                } finally {
+                    pagedInMessagesLock.writeLock().unlock();
                 }
-            } finally {
-                pagedInMessagesLock.writeLock().unlock();
+            } else if (!messages.hasSpace() && isFlowControlLogRequired()) {
+                LOG.warn("{} cursor blocked, no space available to page in messages; usage:
{}", this, this.systemUsage.getMemoryUsage());
             }
         } else {
             // Avoid return null list, if condition is not validated

http://git-wip-us.apache.org/repos/asf/activemq/blob/b29ccf34/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/UsageBlockedDispatchTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/UsageBlockedDispatchTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/UsageBlockedDispatchTest.java
new file mode 100644
index 0000000..7767672
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/UsageBlockedDispatchTest.java
@@ -0,0 +1,156 @@
+/**
+ * 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.ActiveMQConnectionFactory;
+import org.apache.activemq.ActiveMQPrefetchPolicy;
+import org.apache.activemq.TestSupport;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.Queue;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.usage.SystemUsage;
+import org.apache.activemq.util.DefaultTestAppender;
+import org.apache.log4j.Appender;
+import org.apache.log4j.Level;
+import org.apache.log4j.spi.LoggingEvent;
+
+import javax.jms.*;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class UsageBlockedDispatchTest extends TestSupport {
+
+    private static final int MESSAGES_COUNT = 100;
+    private static byte[] buf = new byte[2 * 1024];
+    private BrokerService broker;
+
+    protected long messageReceiveTimeout = 4000L;
+
+    private String connectionUri;
+
+    @Override
+    public void setUp() throws Exception {
+
+        broker = new BrokerService();
+        broker.setDataDirectory("target" + File.separator + "activemq-data");
+        broker.setPersistent(true);
+        broker.setUseJmx(true);
+        broker.setAdvisorySupport(false);
+        broker.setDeleteAllMessagesOnStartup(true);
+
+        setDefaultPersistenceAdapter(broker);
+        SystemUsage sysUsage = broker.getSystemUsage();
+        sysUsage.getMemoryUsage().setLimit(100*1024);
+
+        PolicyEntry defaultPolicy = new PolicyEntry();
+        defaultPolicy.setProducerFlowControl(false);
+        defaultPolicy.setCursorMemoryHighWaterMark(100);
+        defaultPolicy.setMemoryLimit(50*1024);
+
+        PolicyMap policyMap = new PolicyMap();
+        policyMap.setDefaultEntry(defaultPolicy);
+        broker.setDestinationPolicy(policyMap);
+        broker.setSystemUsage(sysUsage);
+
+        broker.addConnector("tcp://localhost:0").setName("Default");
+        broker.start();
+
+        connectionUri = broker.getTransportConnectors().get(0).getPublishableConnectString();
+    }
+
+    @Override
+    public void tearDown() throws Exception {
+        if (broker != null) {
+            broker.stop();
+        }
+    }
+
+    public void testFillMemToBlockConsumer() throws Exception {
+
+        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(connectionUri);
+        ActiveMQPrefetchPolicy prefetch = new ActiveMQPrefetchPolicy();
+        prefetch.setTopicPrefetch(10);
+        factory.setPrefetchPolicy(prefetch);
+
+        final Connection producerConnection = factory.createConnection();
+        producerConnection.start();
+
+        Session producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        MessageProducer producer = producerSession.createProducer(null);
+        BytesMessage message = producerSession.createBytesMessage();
+        message.writeBytes(buf);
+
+        int numFillers = 4;
+        ArrayList<ActiveMQQueue> fillers = new ArrayList<ActiveMQQueue>();
+        for (int i=0; i<numFillers; i++) {
+            fillers.add(new ActiveMQQueue("Q" + i));
+        }
+
+        // fill cache and consume all memory
+        for (int idx = 0; idx < MESSAGES_COUNT; ++idx) {
+            for (ActiveMQQueue q : fillers) {
+                producer.send(q, message);
+            }
+        }
+        ActiveMQQueue willGetAPage = new ActiveMQQueue("Q" + numFillers++);
+        for (int idx = 0; idx < MESSAGES_COUNT; ++idx) {
+            producer.send(willGetAPage, message);
+        }
+
+        ActiveMQQueue shouldBeStuckForDispatch = new ActiveMQQueue("Q" + numFillers);
+        for (int idx = 0; idx < MESSAGES_COUNT; ++idx) {
+            producer.send(shouldBeStuckForDispatch, message);
+        }
+
+        Connection consumerConnection = factory.createConnection();
+        consumerConnection.start();
+
+        Session consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        MessageConsumer consumer = consumerSession.createConsumer(willGetAPage);
+
+        Message m = consumer.receive(messageReceiveTimeout);
+        assertNotNull("got a message", m);
+
+        final AtomicBoolean gotExpectedLogEvent = new AtomicBoolean(false);
+        Appender appender = new DefaultTestAppender() {
+
+            @Override
+            public void doAppend(LoggingEvent event) {
+                if (event.getLevel() == Level.WARN && event.getRenderedMessage().contains("cursor
blocked")) {
+                    gotExpectedLogEvent.set(true);
+                }
+            }
+        };
+
+        try {
+            org.apache.log4j.Logger.getLogger(Queue.class).addAppender(appender);
+
+            MessageConsumer noDispatchConsumer = consumerSession.createConsumer(shouldBeStuckForDispatch);
+
+            m = noDispatchConsumer.receive(messageReceiveTimeout);
+            assertNull("did not get a message", m);
+
+            assertTrue("Got the new warning about the blocked cursor", gotExpectedLogEvent.get());
+        } finally {
+            org.apache.log4j.Logger.getLogger(Queue.class).removeAppender(appender);
+            org.apache.log4j.Logger.getRootLogger().removeAppender(appender);
+        }
+    }
+}


Mime
View raw message