Mailing-List: contact commits-help@qpid.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@qpid.apache.org
Content-Type: text/plain; charset="utf-8"
MIME-Version: 1.0
Content-Transfer-Encoding: 7bit
Subject: svn commit: r1556096 - in /qpid/trunk/qpid/java:
 broker-core/src/main/java/org/apache/qpid/server/queue/
 broker-core/src/main/java/org/apache/qpid/server/subscription/
 systests/src/main/java/org/apache/qpid/server/queue/
Date: Tue, 07 Jan 2014 00:57:57 -0000
To: commits@qpid.apache.org
From: robbie@apache.org
Message-Id: <20140107005758.2F12323889EB@eris.apache.org>

Author: robbie
Date: Tue Jan  7 00:57:57 2014
New Revision: 1556096

URL: http://svn.apache.org/r1556096
Log:
QPID-5450: have the group manager try to acquire the message at the time it is accepted into the group, so that the associated shared-group state change occurs within the single synchronization block

Modified:
    qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
    qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/AssignedSubscriptionMessageGroupManager.java
    qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/DefinedGroupMessageGroupManager.java
    qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MessageGroupQueueTest.java

Modified: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java?rev=1556096&r1=1556095&r2=1556096&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java (original)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java Tue Jan  7 00:57:57 2014
@@ -733,7 +733,7 @@ public class SimpleAMQQueue implements A
                     && mightAssign(sub, entry)
                     && !sub.wouldSuspend(entry))
                 {
-                    if (sub.acquires() && !(assign(sub, entry) && entry.acquire(sub)))
+                    if (sub.acquires() && !assign(sub, entry))
                     {
                         // restore credit here that would have been taken away by wouldSuspend since we didn't manage
                         // to acquire the entry for this subscription
@@ -754,10 +754,18 @@ public class SimpleAMQQueue implements A
 
     private boolean assign(final Subscription sub, final QueueEntry entry)
     {
-        return _messageGroupManager == null || _messageGroupManager.acceptMessage(sub, entry);
+        if(_messageGroupManager == null)
+        {
+            //no grouping, try to acquire immediately.
+            return entry.acquire(sub);
+        }
+        else
+        {
+            //the group manager is responsible for acquiring the message if/when appropriate
+            return _messageGroupManager.acceptMessage(sub, entry);
+        }
     }
 
-
     private boolean mightAssign(final Subscription sub, final QueueEntry entry)
     {
         if(_messageGroupManager == null || !sub.acquires())
@@ -1645,7 +1653,7 @@ public class SimpleAMQQueue implements A
                 {
                     if (!sub.wouldSuspend(node))
                     {
-                        if (sub.acquires() && !(assign(sub, node) && node.acquire(sub)))
+                        if (sub.acquires() && !assign(sub, node))
                         {
                             // restore credit here that would have been taken away by wouldSuspend since we didn't manage
                             // to acquire the entry for this subscription

Modified: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/AssignedSubscriptionMessageGroupManager.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/AssignedSubscriptionMessageGroupManager.java?rev=1556096&r1=1556095&r2=1556096&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/AssignedSubscriptionMessageGroupManager.java (original)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/AssignedSubscriptionMessageGroupManager.java Tue Jan  7 00:57:57 2014
@@ -63,6 +63,18 @@ public class AssignedSubscriptionMessage
 
     public boolean acceptMessage(Subscription sub, QueueEntry entry)
     {
+        if(assignMessage(sub, entry))
+        {
+            return entry.acquire(sub);
+        }
+        else
+        {
+            return false;
+        }
+    }
+
+    private boolean assignMessage(Subscription sub, QueueEntry entry)
+    {
         Object groupVal = entry.getMessage().getMessageHeader().getHeader(_groupId);
         if(groupVal == null)
         {

Modified: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/DefinedGroupMessageGroupManager.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/DefinedGroupMessageGroupManager.java?rev=1556096&r1=1556095&r2=1556096&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/DefinedGroupMessageGroupManager.java (original)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/DefinedGroupMessageGroupManager.java Tue Jan  7 00:57:57 2014
@@ -136,9 +136,21 @@ public class DefinedGroupMessageGroupMan
 
     public synchronized boolean acceptMessage(final Subscription sub, final QueueEntry entry)
     {
+        if(assignMessage(sub, entry))
+        {
+            return entry.acquire(sub);
+        }
+        else
+        {
+            return false;
+        }
+    }
+
+    private boolean assignMessage(final Subscription sub, final QueueEntry entry)
+    {
         Object groupId = getKey(entry);
         Group group = _groupMap.get(groupId);
-        
+
         if(group == null || !group.isValid())
         {
             group = new Group(groupId, sub);
@@ -152,11 +164,10 @@ public class DefinedGroupMessageGroupMan
             {
                 return false;
             }
-
         }
-        
+
         Subscription assignedSub = group.getSubscription();
-        
+
         if(assignedSub == sub)
         {
             entry.addStateChangeListener(new GroupStateChangeListener(group, entry));
@@ -167,8 +178,7 @@ public class DefinedGroupMessageGroupMan
             return false;            
         }
     }
-    
-    
+
     public synchronized QueueEntry findEarliestAssignedAvailableEntry(final Subscription sub)
     {
         EntryFinder visitor = new EntryFinder(sub);

Modified: qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MessageGroupQueueTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MessageGroupQueueTest.java?rev=1556096&r1=1556095&r2=1556096&view=diff
==============================================================================
--- qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MessageGroupQueueTest.java (original)
+++ qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MessageGroupQueueTest.java Tue Jan  7 00:57:57 2014
@@ -20,22 +20,28 @@
 */
 package org.apache.qpid.server.queue;
 
-import org.apache.qpid.AMQException;
-import org.apache.qpid.client.AMQConnection;
-import org.apache.qpid.client.AMQDestination;
-import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.test.utils.QpidBrokerTestCase;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.jms.Connection;
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
 import javax.jms.MessageProducer;
 import javax.jms.Queue;
 import javax.jms.Session;
-import java.util.HashMap;
-import java.util.Map;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.AMQDestination;
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.jms.ConnectionURL;
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
 
 public class MessageGroupQueueTest extends QpidBrokerTestCase
 {
@@ -469,7 +475,6 @@ public class MessageGroupQueueTest exten
 
     }
 
-
     private Message createMessage(int msg, String group) throws JMSException
     {
         Message send = producerSession.createTextMessage("Message: " + msg);
@@ -478,4 +483,122 @@ public class MessageGroupQueueTest exten
 
         return send;
     }
+
+    /**
+     * Tests that when a number of new messages for a given groupid are arriving while the delivery group
+     * state is also in the process of being emptied (due to acking a message while using prefetch=1), that only
+     * 1 of a number of existing consumers is ever receiving messages for the shared group at a time.
+     */
+    public void testSingleSharedGroupWithMultipleConsumers() throws Exception
+    {
+        final Map<String,Object> arguments = new HashMap<String, Object>();
+        arguments.put(QueueArgumentsConverter.QPID_GROUP_HEADER_KEY,"group");
+        arguments.put(QueueArgumentsConverter.QPID_SHARED_MSG_GROUP,"1");
+
+        ((AMQSession) producerSession).createQueue(new AMQShortString(QUEUE), true, false, false, arguments);
+        queue = (Queue) producerSession.createQueue("direct://amq.direct/"+QUEUE+"/"+QUEUE+"?durable='false'&autodelete='true'");
+
+        ((AMQSession) producerSession).declareAndBind((AMQDestination)queue);
+        producer = producerSession.createProducer(queue);
+
+
+        consumerConnection.close();
+        Map<String, String> options = new HashMap<String, String>();
+        options.put(ConnectionURL.OPTIONS_MAXPREFETCH, "1");
+        consumerConnection = getConnectionWithOptions(options);
+
+        int numMessages = 100;
+        SharedGroupTestMessageListener groupingTestMessageListener = new SharedGroupTestMessageListener(numMessages);
+
+        Session cs1 = ((AMQConnection)consumerConnection).createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Session cs2 = ((AMQConnection)consumerConnection).createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Session cs3 = ((AMQConnection)consumerConnection).createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Session cs4 = ((AMQConnection)consumerConnection).createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+        MessageConsumer consumer1 = cs1.createConsumer(queue);
+        consumer1.setMessageListener(groupingTestMessageListener);
+        MessageConsumer consumer2 = cs2.createConsumer(queue);
+        consumer2.setMessageListener(groupingTestMessageListener);
+        MessageConsumer consumer3 = cs3.createConsumer(queue);
+        consumer3.setMessageListener(groupingTestMessageListener);
+        MessageConsumer consumer4 = cs4.createConsumer(queue);
+        consumer4.setMessageListener(groupingTestMessageListener);
+        consumerConnection.start();
+
+        for(int i = 1; i <= numMessages; i++)
+        {
+            producer.send(createMessage(i, "GROUP"));
+        }
+        producerSession.commit();
+        producer.close();
+        producerSession.close();
+        producerConnection.close();
+
+        assertTrue("Mesages not all recieved in the allowed timeframe", groupingTestMessageListener.waitForLatch(30));
+        assertEquals("Unexpected concurrent processing of messages for the group", 0, groupingTestMessageListener.getConcurrentProcessingCases());
+        assertNull("Unexpecte throwable in message listeners", groupingTestMessageListener.getThrowable());
+    }
+
+    public static class SharedGroupTestMessageListener implements MessageListener
+    {
+        private final CountDownLatch _count;
+        private final AtomicInteger _activeListeners = new AtomicInteger();
+        private final AtomicInteger _concurrentProcessingCases = new AtomicInteger();
+        private Throwable _throwable;
+
+        public SharedGroupTestMessageListener(int numMessages)
+        {
+            _count = new CountDownLatch(numMessages);
+        }
+
+        public void onMessage(Message message)
+        {
+            try
+            {
+                int currentActiveListeners = _activeListeners.incrementAndGet();
+
+                if (currentActiveListeners > 1)
+                {
+                    _concurrentProcessingCases.incrementAndGet();
+
+                    System.err.println("Concurrent processing when handling message: " + message.getIntProperty("msg"));
+                }
+
+                try
+                {
+                    Thread.sleep(25);
+                }
+                catch (InterruptedException e)
+                {
+                    Thread.currentThread().interrupt();
+                }
+
+                _activeListeners.decrementAndGet();
+            }
+            catch (Throwable t)
+            {
+                _throwable = t;
+                t.printStackTrace();
+            }
+            finally
+            {
+                _count.countDown();
+            }
+        }
+
+        public boolean waitForLatch(int seconds) throws Exception
+        {
+            return _count.await(seconds, TimeUnit.SECONDS);
+        }
+
+        public int getConcurrentProcessingCases()
+        {
+            return _concurrentProcessingCases.get();
+        }
+
+        public Throwable getThrowable()
+        {
+            return _throwable;
+        }
+    }
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org