activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cshan...@apache.org
Subject activemq git commit: https://issues.apache.org/jira/browse/AMQ-5712
Date Fri, 20 Nov 2015 20:58:47 GMT
Repository: activemq
Updated Branches:
  refs/heads/master b5dd0a16f -> cc6213ebf


https://issues.apache.org/jira/browse/AMQ-5712

Switching addMessageLast to tryAddMessageLast when messages are added
to a Queue pending cursor to allow a potential deadlock to be
avoided. There is more work to be done here but this will at least
prevent a deadlock from occurring.

Fix and test based off of a patch created by Timothy Bish.


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

Branch: refs/heads/master
Commit: cc6213ebf25a129b278a2ff0d7c32c25edd71eaa
Parents: b5dd0a1
Author: Christopher L. Shannon (cshannon) <christopher.l.shannon@gmail.com>
Authored: Fri Nov 20 20:45:38 2015 +0000
Committer: Christopher L. Shannon (cshannon) <christopher.l.shannon@gmail.com>
Committed: Fri Nov 20 20:58:27 2015 +0000

----------------------------------------------------------------------
 .../apache/activemq/broker/region/Queue.java    |  72 +++---
 .../cursors/AbstractPendingMessageCursor.java   |  90 ++++++--
 .../region/cursors/AbstractStoreCursor.java     |   4 +-
 .../cursors/FilePendingMessageCursor.java       |   5 -
 .../region/cursors/PendingMessageCursor.java    |   2 +
 .../cursors/StoreDurableSubscriberCursor.java   |   2 +-
 .../broker/region/cursors/StoreQueueCursor.java |   4 +-
 .../region/cursors/VMPendingMessageCursor.java  |   2 +-
 .../org/apache/activemq/bugs/AMQ5712Test.java   | 231 +++++++++++++++++++
 9 files changed, 352 insertions(+), 60 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq/blob/cc6213eb/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 e6c20de..adc3a53 100755
--- 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
@@ -827,33 +827,38 @@ public class Queue extends BaseDestination implements Task, UsageListener,
Index
         ListenableFuture<Object> result = null;
 
         producerExchange.incrementSend();
-        checkUsage(context, producerExchange, message);
-        sendLock.lockInterruptibly();
-        try {
-            message.getMessageId().setBrokerSequenceId(getDestinationSequenceId());
-            if (store != null && message.isPersistent()) {
-                message.getMessageId().setFutureOrSequenceLong(null);
-                try {
-                    if (messages.isCacheEnabled()) {
-                        result = store.asyncAddQueueMessage(context, message, isOptimizeStorage());
-                        result.addListener(new PendingMarshalUsageTracker(message));
-                    } else {
-                        store.addMessage(context, message);
-                    }
-                    if (isReduceMemoryFootprint()) {
-                        message.clearMarshalledState();
+        do {
+            checkUsage(context, producerExchange, message);
+            sendLock.lockInterruptibly();
+            try {
+                message.getMessageId().setBrokerSequenceId(getDestinationSequenceId());
+                if (store != null && message.isPersistent()) {
+                    message.getMessageId().setFutureOrSequenceLong(null);
+                    try {
+                        if (messages.isCacheEnabled()) {
+                            result = store.asyncAddQueueMessage(context, message, isOptimizeStorage());
+                            result.addListener(new PendingMarshalUsageTracker(message));
+                        } else {
+                            store.addMessage(context, message);
+                        }
+                        if (isReduceMemoryFootprint()) {
+                            message.clearMarshalledState();
+                        }
+                    } catch (Exception e) {
+                        // we may have a store in inconsistent state, so reset the cursor
+                        // before restarting normal broker operations
+                        resetNeeded = true;
+                        throw e;
                     }
-                } catch (Exception e) {
-                    // we may have a store in inconsistent state, so reset the cursor
-                    // before restarting normal broker operations
-                    resetNeeded = true;
-                    throw e;
                 }
+                if(tryOrderedCursorAdd(message, context)) {
+                    break;
+                }
+            } finally {
+                sendLock.unlock();
             }
-            orderedCursorAdd(message, context);
-        } finally {
-            sendLock.unlock();
-        }
+        } while (started.get());
+
         if (store == null || (!context.isInTransaction() && !message.isPersistent()))
{
             messageSent(context, message);
         }
@@ -867,15 +872,19 @@ public class Queue extends BaseDestination implements Task, UsageListener,
Index
         }
     }
 
-    private void orderedCursorAdd(Message message, ConnectionContext context) throws Exception
{
+    private boolean tryOrderedCursorAdd(Message message, ConnectionContext context) throws
Exception {
+        boolean result = true;
+
         if (context.isInTransaction()) {
             context.getTransaction().addSynchronization(new CursorAddSync(new MessageContext(context,
message, null)));
         } else if (store != null && message.isPersistent()) {
             doPendingCursorAdditions();
         } else {
             // no ordering issue with non persistent messages
-            cursorAdd(message);
+            result = tryCursorAdd(message);
         }
+
+        return result;
     }
 
     private void checkUsage(ConnectionContext context,ProducerBrokerExchange producerBrokerExchange,
Message message) throws ResourceAllocationException, IOException, InterruptedException {
@@ -1813,7 +1822,7 @@ public class Queue extends BaseDestination implements Task, UsageListener,
Index
         }
     }
 
-    final boolean cursorAdd(final Message msg) throws Exception {
+    private final boolean cursorAdd(final Message msg) throws Exception {
         messagesLock.writeLock().lock();
         try {
             return messages.addMessageLast(msg);
@@ -1822,6 +1831,15 @@ public class Queue extends BaseDestination implements Task, UsageListener,
Index
         }
     }
 
+    private final boolean tryCursorAdd(final Message msg) throws Exception {
+        messagesLock.writeLock().lock();
+        try {
+            return messages.tryAddMessageLast(msg, 50);
+        } finally {
+            messagesLock.writeLock().unlock();
+        }
+    }
+
     final void messageSent(final ConnectionContext context, final Message msg) throws Exception
{
         destinationStatistics.getEnqueues().increment();
         destinationStatistics.getMessages().increment();

http://git-wip-us.apache.org/repos/asf/activemq/blob/cc6213eb/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractPendingMessageCursor.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractPendingMessageCursor.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractPendingMessageCursor.java
index 12ea104..0857482 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractPendingMessageCursor.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/AbstractPendingMessageCursor.java
@@ -33,8 +33,8 @@ import org.apache.activemq.usage.SystemUsage;
 /**
  * Abstract method holder for pending message (messages awaiting disptach to a
  * consumer) cursor
- * 
- * 
+ *
+ *
  */
 public abstract class AbstractPendingMessageCursor implements PendingMessageCursor {
     protected int memoryUsageHighWaterMark = 70;
@@ -49,12 +49,13 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
     private boolean started=false;
     protected MessageReference last = null;
     protected final boolean prioritizedMessages;
-    
+
     public AbstractPendingMessageCursor(boolean prioritizedMessages) {
         this.prioritizedMessages=prioritizedMessages;
     }
-  
 
+
+    @Override
     public synchronized void start() throws Exception  {
         if (!started && enableAudit && audit==null) {
             audit= new ActiveMQMessageAudit(maxAuditDepth,maxProducersToAudit);
@@ -62,71 +63,89 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
         started=true;
     }
 
+    @Override
     public synchronized void stop() throws Exception  {
         started=false;
         gc();
     }
 
+    @Override
     public void add(ConnectionContext context, Destination destination) throws Exception
{
     }
 
+    @Override
     @SuppressWarnings("unchecked")
     public List<MessageReference> remove(ConnectionContext context, Destination destination)
throws Exception {
         return Collections.EMPTY_LIST;
     }
 
+    @Override
     public boolean isRecoveryRequired() {
         return true;
     }
 
+    @Override
     public void addMessageFirst(MessageReference node) throws Exception {
     }
 
+    @Override
     public boolean addMessageLast(MessageReference node) throws Exception {
-        return true;
+        return tryAddMessageLast(node, INFINITE_WAIT);
     }
-    
+
+    @Override
     public boolean tryAddMessageLast(MessageReference node, long maxWaitTime) throws Exception
{
-        return addMessageLast(node);
+        return true;
     }
 
+    @Override
     public void addRecoveredMessage(MessageReference node) throws Exception {
         addMessageLast(node);
     }
 
+    @Override
     public void clear() {
     }
 
+    @Override
     public boolean hasNext() {
         return false;
     }
 
+    @Override
     public boolean isEmpty() {
         return false;
     }
 
+    @Override
     public boolean isEmpty(Destination destination) {
         return isEmpty();
     }
 
+    @Override
     public MessageReference next() {
         return null;
     }
 
+    @Override
     public void remove() {
     }
 
+    @Override
     public void reset() {
     }
 
+    @Override
     public int size() {
         return 0;
     }
 
+    @Override
     public int getMaxBatchSize() {
         return maxBatchSize;
     }
 
+    @Override
     public void setMaxBatchSize(int maxBatchSize) {
         this.maxBatchSize = maxBatchSize;
     }
@@ -134,31 +153,39 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
     protected void fillBatch() throws Exception {
     }
 
+    @Override
     public void resetForGC() {
         reset();
     }
 
+    @Override
     public void remove(MessageReference node) {
     }
 
+    @Override
     public void gc() {
     }
 
+    @Override
     public void setSystemUsage(SystemUsage usageManager) {
         this.systemUsage = usageManager;
     }
 
+    @Override
     public boolean hasSpace() {
         return systemUsage != null ? (!systemUsage.getMemoryUsage().isFull(memoryUsageHighWaterMark))
: true;
     }
 
+    @Override
     public boolean isFull() {
         return systemUsage != null ? systemUsage.getMemoryUsage().isFull() : false;
     }
 
+    @Override
     public void release() {
     }
 
+    @Override
     public boolean hasMessagesBufferedToDeliver() {
         return false;
     }
@@ -166,6 +193,7 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
     /**
      * @return the memoryUsageHighWaterMark
      */
+    @Override
     public int getMemoryUsageHighWaterMark() {
         return memoryUsageHighWaterMark;
     }
@@ -173,6 +201,7 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
     /**
      * @param memoryUsageHighWaterMark the memoryUsageHighWaterMark to set
      */
+    @Override
     public void setMemoryUsageHighWaterMark(int memoryUsageHighWaterMark) {
         this.memoryUsageHighWaterMark = memoryUsageHighWaterMark;
     }
@@ -180,25 +209,28 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
     /**
      * @return the usageManager
      */
+    @Override
     public SystemUsage getSystemUsage() {
         return this.systemUsage;
     }
 
     /**
      * destroy the cursor
-     * 
+     *
      * @throws Exception
      */
+    @Override
     public void destroy() throws Exception {
         stop();
     }
 
     /**
      * Page in a restricted number of messages
-     * 
+     *
      * @param maxItems maximum number of messages to return
      * @return a list of paged in messages
      */
+    @Override
     public LinkedList<MessageReference> pageInList(int maxItems) {
         throw new RuntimeException("Not supported");
     }
@@ -206,6 +238,7 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
     /**
      * @return the maxProducersToAudit
      */
+    @Override
     public int getMaxProducersToAudit() {
         return maxProducersToAudit;
     }
@@ -213,6 +246,7 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
     /**
      * @param maxProducersToAudit the maxProducersToAudit to set
      */
+    @Override
     public synchronized void setMaxProducersToAudit(int maxProducersToAudit) {
         this.maxProducersToAudit = maxProducersToAudit;
         if (audit != null) {
@@ -223,25 +257,28 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
     /**
      * @return the maxAuditDepth
      */
+    @Override
     public int getMaxAuditDepth() {
         return maxAuditDepth;
     }
-    
+
 
     /**
      * @param maxAuditDepth the maxAuditDepth to set
      */
+    @Override
     public synchronized void setMaxAuditDepth(int maxAuditDepth) {
         this.maxAuditDepth = maxAuditDepth;
         if (audit != null) {
             audit.setAuditDepth(maxAuditDepth);
         }
     }
-    
-    
+
+
     /**
      * @return the enableAudit
      */
+    @Override
     public boolean isEnableAudit() {
         return enableAudit;
     }
@@ -249,38 +286,44 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
     /**
      * @param enableAudit the enableAudit to set
      */
+    @Override
     public synchronized void setEnableAudit(boolean enableAudit) {
         this.enableAudit = enableAudit;
         if (enableAudit && started && audit==null) {
             audit= new ActiveMQMessageAudit(maxAuditDepth,maxProducersToAudit);
         }
     }
-    
+
+    @Override
     public boolean isTransient() {
         return false;
     }
-    
-       
+
+
     /**
      * set the audit
      * @param audit new audit component
      */
+    @Override
     public void setMessageAudit(ActiveMQMessageAudit audit) {
     	this.audit=audit;
     }
-    
-    
+
+
     /**
      * @return the audit
      */
+    @Override
     public ActiveMQMessageAudit getMessageAudit() {
     	return audit;
     }
-    
+
+    @Override
     public boolean isUseCache() {
         return useCache;
     }
 
+    @Override
     public void setUseCache(boolean useCache) {
         this.useCache = useCache;
     }
@@ -290,7 +333,7 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
         rollback(messageId);
         return !unique;
     }
-    
+
     /**
      * records a message id and checks if it is a duplicate
      * @param messageId
@@ -302,17 +345,18 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
         }
         return !audit.isDuplicate(messageId);
     }
-    
+
+    @Override
     public synchronized void rollback(MessageId id) {
         if (audit != null) {
             audit.rollback(id);
         }
     }
-    
+
     public synchronized boolean isStarted() {
         return started;
     }
-    
+
     public static boolean isPrioritizedMessageSubscriber(Broker broker,Subscription sub)
{
         boolean result = false;
         Set<Destination> destinations = broker.getDestinations(sub.getActiveMQDestination());
@@ -328,6 +372,7 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
 
     }
 
+    @Override
     public synchronized boolean isCacheEnabled() {
         return cacheEnabled;
     }
@@ -336,6 +381,7 @@ public abstract class AbstractPendingMessageCursor implements PendingMessageCurs
         cacheEnabled = val;
     }
 
+    @Override
     public void rebase() {
     }
 }

http://git-wip-us.apache.org/repos/asf/activemq/blob/cc6213eb/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 05e4b1f..c6cca59 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
@@ -44,7 +44,7 @@ public abstract class AbstractStoreCursor extends AbstractPendingMessageCursor
i
     private Iterator<MessageReference> iterator = null;
     protected boolean batchResetNeeded = false;
     protected int size;
-    private LinkedList<MessageId> pendingCachedIds = new LinkedList<>();
+    private final LinkedList<MessageId> pendingCachedIds = new LinkedList<>();
     private static int SYNC_ADD = 0;
     private static int ASYNC_ADD = 1;
     final MessageId[] lastCachedIds = new MessageId[2];
@@ -210,7 +210,7 @@ public abstract class AbstractStoreCursor extends AbstractPendingMessageCursor
i
     }
 
     @Override
-    public synchronized boolean addMessageLast(MessageReference node) throws Exception {
+    public synchronized boolean tryAddMessageLast(MessageReference node, long wait) throws
Exception {
         boolean disableCache = false;
         if (hasSpace()) {
             if (!isCacheEnabled() && size==0 && isStarted() && useCache)
{

http://git-wip-us.apache.org/repos/asf/activemq/blob/cc6213eb/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
index 3f3f33b..9c9a8e7 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
@@ -203,11 +203,6 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor
imple
      * @throws Exception
      */
     @Override
-    public synchronized boolean addMessageLast(MessageReference node) throws Exception {
-        return tryAddMessageLast(node, 0);
-    }
-
-    @Override
     public synchronized boolean tryAddMessageLast(MessageReference node, long maxWaitTime)
throws Exception {
         if (!node.isExpired()) {
             try {

http://git-wip-us.apache.org/repos/asf/activemq/blob/cc6213eb/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/PendingMessageCursor.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/PendingMessageCursor.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/PendingMessageCursor.java
index bf7fd7a..6359635 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/PendingMessageCursor.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/PendingMessageCursor.java
@@ -35,6 +35,8 @@ import org.apache.activemq.usage.SystemUsage;
  */
 public interface PendingMessageCursor extends Service {
 
+    static final long INFINITE_WAIT = 0;
+
     /**
      * Add a destination
      *

http://git-wip-us.apache.org/repos/asf/activemq/blob/cc6213eb/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java
index 9d723b8..269bde3 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java
@@ -183,7 +183,7 @@ public class StoreDurableSubscriberCursor extends AbstractPendingMessageCursor
{
     }
 
     @Override
-    public synchronized boolean addMessageLast(MessageReference node) throws Exception {
+    public synchronized boolean tryAddMessageLast(MessageReference node, long wait) throws
Exception {
         if (node != null) {
             Message msg = node.getMessage();
             if (isStarted()) {

http://git-wip-us.apache.org/repos/asf/activemq/blob/cc6213eb/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/StoreQueueCursor.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/StoreQueueCursor.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/StoreQueueCursor.java
index caa93b6..7f26b43 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/StoreQueueCursor.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/StoreQueueCursor.java
@@ -90,14 +90,14 @@ public class StoreQueueCursor extends AbstractPendingMessageCursor {
     }
 
     @Override
-    public synchronized boolean addMessageLast(MessageReference node) throws Exception {
+    public synchronized boolean tryAddMessageLast(MessageReference node, long maxWait) throws
Exception {
         boolean result = true;
         if (node != null) {
             Message msg = node.getMessage();
             if (started) {
                 pendingCount++;
                 if (!msg.isPersistent()) {
-                    nonPersistent.addMessageLast(node);
+                    result = nonPersistent.tryAddMessageLast(node, maxWait);
                 }
             }
             if (msg.isPersistent()) {

http://git-wip-us.apache.org/repos/asf/activemq/blob/cc6213eb/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/VMPendingMessageCursor.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/VMPendingMessageCursor.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/VMPendingMessageCursor.java
index 75be766..cd4da9d 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/VMPendingMessageCursor.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/VMPendingMessageCursor.java
@@ -102,7 +102,7 @@ public class VMPendingMessageCursor extends AbstractPendingMessageCursor
{
      */
 
     @Override
-    public synchronized boolean addMessageLast(MessageReference node) {
+    public synchronized boolean tryAddMessageLast(MessageReference node, long maxWait) {
         node.incrementReferenceCount();
         list.addMessageLast(node);
         return true;

http://git-wip-us.apache.org/repos/asf/activemq/blob/cc6213eb/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ5712Test.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ5712Test.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ5712Test.java
new file mode 100644
index 0000000..4a396a4
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ5712Test.java
@@ -0,0 +1,231 @@
+/**
+ * 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.bugs;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.jms.BytesMessage;
+import javax.jms.Connection;
+import javax.jms.DeliveryMode;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.management.MalformedObjectNameException;
+import javax.management.ObjectName;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.jmx.QueueViewMBean;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.store.kahadb.KahaDBStore;
+import org.apache.activemq.store.kahadb.plist.PListStoreImpl;
+import org.apache.activemq.util.Wait;
+import org.apache.activemq.util.Wait.Condition;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test behavior of senders when broker side producer flow control kicks in.
+ */
+public class AMQ5712Test {
+
+    private static final Logger LOG = LoggerFactory.getLogger(AMQ5712Test.class);
+
+    @Rule public TestName name = new TestName();
+
+    private BrokerService brokerService;
+    private Connection connection;
+
+    @Before
+    public void setUp() throws Exception {
+        brokerService = createBroker();
+        brokerService.start();
+        brokerService.waitUntilStarted();
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        if (connection != null) {
+            try {
+                connection.close();
+            } catch (Exception e) {}
+        }
+
+        if (brokerService != null) {
+            brokerService.stop();
+            brokerService.waitUntilStopped();
+            brokerService = null;
+        }
+    }
+
+    private Connection createConnection() throws Exception {
+        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost?create=false");
+        factory.setAlwaysSyncSend(true);
+        return factory.createConnection();
+    }
+
+    @Test(timeout = 120000)
+    public void test() throws Exception {
+        connection = createConnection();
+        connection.start();
+
+        final int MSG_COUNT = 100;
+
+        final Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+        final Queue queue = session.createQueue(name.getMethodName());
+        MessageProducer producer = session.createProducer(queue);
+        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+
+        final QueueViewMBean queueView = getProxyToQueue(name.getMethodName());
+
+        byte[] payload = new byte[65535];
+        Arrays.fill(payload, (byte) 255);
+        final CountDownLatch done = new CountDownLatch(1);
+        final AtomicInteger counter = new AtomicInteger();
+
+        Thread purge = new Thread(new Runnable() {
+
+            @Override
+            public void run() {
+                try {
+                    while (!done.await(5, TimeUnit.SECONDS)) {
+                        if (queueView.getBlockedSends() > 0 && queueView.getQueueSize()
> 0) {
+                            long queueSize = queueView.getQueueSize();
+                            LOG.info("Queue send blocked at {} messages", queueSize);
+                            MessageConsumer consumer = session.createConsumer(queue);
+                            for (int i = 0; i < queueSize; i++) {
+                                Message message = consumer.receive(60000);
+                                if (message != null) {
+                                    counter.incrementAndGet();
+                                    message.acknowledge();
+                                } else {
+                                    LOG.warn("Got null message when none as expected.");
+                                }
+                            }
+                            consumer.close();
+                        }
+                    }
+                } catch (Exception ex) {
+                }
+            }
+        });
+        purge.start();
+
+        for (int i = 0; i < MSG_COUNT; i++) {
+            BytesMessage message = session.createBytesMessage();
+            message.writeBytes(payload);
+            producer.send(message);
+            LOG.info("sent message: {}", i);
+        }
+
+        done.countDown();
+        purge.join(60000);
+        if (purge.isAlive()) {
+            fail("Consumer thread should have read initial batch and completed.");
+        }
+
+        //wait for processed acked messages
+        assertTrue(Wait.waitFor(new Condition() {
+            @Override
+            public boolean isSatisified() throws Exception {
+                return queueView.getDequeueCount() == counter.get();
+            }
+        }));
+
+        long remainingQueued = queueView.getQueueSize();
+        LOG.info("Remaining messages to consume: {}", remainingQueued);
+        assertEquals(remainingQueued, MSG_COUNT - counter.get());
+
+        MessageConsumer consumer = session.createConsumer(queue);
+        for (int i = counter.get(); i < MSG_COUNT; i++) {
+            Message message = consumer.receive(5000);
+            assertNotNull("Should not get null message", consumer);
+            counter.incrementAndGet();
+            message.acknowledge();
+            LOG.info("Read message: {}", i);
+        }
+
+        assertEquals("Should consume all messages", MSG_COUNT, counter.get());
+    }
+
+    protected BrokerService createBroker() throws Exception {
+        BrokerService answer = new BrokerService();
+
+        KahaDBStore persistence = createStore(true);
+        persistence.setJournalMaxFileLength(1024 * 1024 * 1);
+
+        answer.setPersistent(true);
+        answer.setPersistenceAdapter(persistence);
+        answer.setDeleteAllMessagesOnStartup(true);
+        answer.getSystemUsage().getMemoryUsage().setLimit(1024 * 1024 * 6);
+        answer.getSystemUsage().getTempUsage().setLimit(1024 * 1024 * 5);
+        answer.getSystemUsage().getStoreUsage().setLimit(1024 * 1024 * 5);
+        answer.setUseJmx(true);
+        answer.getManagementContext().setCreateConnector(false);
+        answer.setSchedulerSupport(false);
+        answer.setAdvisorySupport(false);
+
+        PListStoreImpl tempStore = ((PListStoreImpl)answer.getSystemUsage().getTempUsage().getStore());
+        tempStore.setCleanupInterval(10000);
+        tempStore.setJournalMaxFileLength(1024 * 1024 * 2);
+
+        PolicyEntry policy = new PolicyEntry();
+        policy.setProducerFlowControl(false);
+
+        PolicyMap policyMap = new PolicyMap();
+        policyMap.setDefaultEntry(policy);
+
+        answer.setDestinationPolicy(policyMap);
+
+        return answer;
+    }
+
+    private KahaDBStore createStore(boolean delete) throws IOException {
+        KahaDBStore kaha = new KahaDBStore();
+        kaha.setDirectory(new File("target/activemq-data/kahadb"));
+        if( delete ) {
+            kaha.deleteAllMessages();
+        }
+        return kaha;
+    }
+
+    protected QueueViewMBean getProxyToQueue(String name) throws MalformedObjectNameException,
JMSException {
+        ObjectName queueViewMBeanName = new ObjectName("org.apache.activemq:type=Broker,brokerName=localhost,destinationType=Queue,destinationName="+name);
+        QueueViewMBean proxy = (QueueViewMBean) brokerService.getManagementContext()
+                .newProxyInstance(queueViewMBeanName, QueueViewMBean.class, true);
+        return proxy;
+    }
+}
\ No newline at end of file


Mime
View raw message