From cddbd677ec47e459f76b83825aa98d3c9ab782dc Mon Sep 17 00:00:00 2001 From: Robert Gemmell Date: Tue, 7 Jan 2014 00:57:57 +0000 Subject: 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 git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1556096 13f79535-47bb-0310-9956-ffa450edef68 --- .../apache/qpid/server/queue/SimpleAMQQueue.java | 16 ++- .../AssignedSubscriptionMessageGroupManager.java | 12 ++ .../DefinedGroupMessageGroupManager.java | 22 +++- .../qpid/server/queue/MessageGroupQueueTest.java | 141 +++++++++++++++++++-- 4 files changed, 172 insertions(+), 19 deletions(-) (limited to 'qpid/java') diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java index 2aa3fd6ce8..b002419064 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java @@ -733,7 +733,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes && 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 AMQQueue, Subscription.StateListener, Mes 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 AMQQueue, Subscription.StateListener, Mes { 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 diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/AssignedSubscriptionMessageGroupManager.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/AssignedSubscriptionMessageGroupManager.java index efedad1181..ae7e11afa4 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/AssignedSubscriptionMessageGroupManager.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/AssignedSubscriptionMessageGroupManager.java @@ -62,6 +62,18 @@ public class AssignedSubscriptionMessageGroupManager implements MessageGroupMana } 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) diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/DefinedGroupMessageGroupManager.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/DefinedGroupMessageGroupManager.java index f38e23b342..55110c46de 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/DefinedGroupMessageGroupManager.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/DefinedGroupMessageGroupManager.java @@ -135,10 +135,22 @@ public class DefinedGroupMessageGroupManager implements MessageGroupManager } 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 DefinedGroupMessageGroupManager implements MessageGroupManager { return false; } - } - + Subscription assignedSub = group.getSubscription(); - + if(assignedSub == sub) { entry.addStateChangeListener(new GroupStateChangeListener(group, entry)); @@ -167,8 +178,7 @@ public class DefinedGroupMessageGroupManager implements MessageGroupManager return false; } } - - + public synchronized QueueEntry findEarliestAssignedAvailableEntry(final Subscription sub) { EntryFinder visitor = new EntryFinder(sub); diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MessageGroupQueueTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MessageGroupQueueTest.java index d5cee17fa8..cb8ced4ddb 100644 --- a/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MessageGroupQueueTest.java +++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/MessageGroupQueueTest.java @@ -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 extends QpidBrokerTestCase } - private Message createMessage(int msg, String group) throws JMSException { Message send = producerSession.createTextMessage("Message: " + msg); @@ -478,4 +483,122 @@ public class MessageGroupQueueTest extends QpidBrokerTestCase 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 arguments = new HashMap(); + 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 options = new HashMap(); + 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; + } + } } -- cgit v1.2.1