From 226d12f87ca09c06eddb9fb98d6b872a2dfe804f Mon Sep 17 00:00:00 2001
From: Francesco Nigro <nigro.fra@gmail.com>
Date: Tue, 31 Jul 2018 11:16:26 +0200
Subject: [PATCH] ARTEMIS-1999 Broker uses 100% core's CPU time if msg grouping
 is used

The deliver loop won't give up trying to deliver messages when
back-pressure kicks in (credits and/or TCP) if msg grouping is used and
there are many consumers registered: this change will allow the loop
to exit by instructing the logic that the group consumer is the only
consumer to check.
---
 .../artemis/core/server/impl/QueueImpl.java   |  6 +-
 .../core/server/impl/QueueImplTest.java       | 60 +++++++++++++++++++
 2 files changed, 63 insertions(+), 3 deletions(-)

diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
index 265621741da1..c523aa84c997 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
@@ -2484,10 +2484,10 @@ private void deliver() {
                }
             }
 
-            if (pos == endPos) {
-               // Round robin'd all
+            if (pos == endPos || groupConsumer != null) {
+               // Round robin'd all or groupConsumer is defined
 
-               if (noDelivery == size) {
+               if (noDelivery == size || (groupConsumer != null && noDelivery > 0)) {
                   if (handledconsumer != null) {
                      // this shouldn't really happen,
                      // however I'm keeping this as an assertion case future developers ever change the logic here on this class
diff --git a/tests/timing-tests/src/test/java/org/apache/activemq/artemis/tests/timing/core/server/impl/QueueImplTest.java b/tests/timing-tests/src/test/java/org/apache/activemq/artemis/tests/timing/core/server/impl/QueueImplTest.java
index 0181a05b4364..dabdd5a818cb 100644
--- a/tests/timing-tests/src/test/java/org/apache/activemq/artemis/tests/timing/core/server/impl/QueueImplTest.java
+++ b/tests/timing-tests/src/test/java/org/apache/activemq/artemis/tests/timing/core/server/impl/QueueImplTest.java
@@ -25,6 +25,7 @@
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.activemq.artemis.api.core.Message;
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.core.server.Consumer;
 import org.apache.activemq.artemis.core.server.HandleStatus;
@@ -245,4 +246,63 @@ public void disconnect() {
       Assert.assertTrue(gotLatch);
    }
 
+   @Test
+   public void testGroupMessageWithManyConsumers() throws Exception {
+      final CountDownLatch firstMessageHandled = new CountDownLatch(1);
+      final CountDownLatch finished = new CountDownLatch(2);
+      final Consumer groupConsumer = new FakeConsumer() {
+
+         int count = 0;
+
+         @Override
+         public synchronized HandleStatus handle(MessageReference reference) {
+            if (count == 0) {
+               //the first message is handled and will be used to determine this consumer
+               //to be the group consumer
+               count++;
+               firstMessageHandled.countDown();
+               return HandleStatus.HANDLED;
+            } else if (count <= 2) {
+               //the next two attempts to send the second message will be done
+               //attempting a direct delivery and an async one after that
+               count++;
+               finished.countDown();
+               return HandleStatus.BUSY;
+            } else {
+               //this shouldn't happen, because the last attempt to deliver
+               //the second message should have stop the delivery loop:
+               //it will succeed just to let the message being handled and
+               //reduce the message count to 0
+               return HandleStatus.HANDLED;
+            }
+         }
+      };
+      final Consumer noConsumer = new FakeConsumer() {
+         @Override
+         public synchronized HandleStatus handle(MessageReference reference) {
+            Assert.fail("this consumer isn't allowed to consume any message");
+            throw new AssertionError();
+         }
+      };
+      final QueueImpl queue = new QueueImpl(1, new SimpleString("address1"), QueueImplTest.queue1,
+                                            null, null, false, true, false,
+                                            scheduledExecutor, null, null, null,
+                                            ArtemisExecutor.delegate(executor), null, null);
+      queue.addConsumer(groupConsumer);
+      queue.addConsumer(noConsumer);
+      final MessageReference firstMessageReference = generateReference(queue, 1);
+      final SimpleString groupName = SimpleString.toSimpleString("group");
+      firstMessageReference.getMessage().putStringProperty(Message.HDR_GROUP_ID, groupName);
+      final MessageReference secondMessageReference = generateReference(queue, 2);
+      secondMessageReference.getMessage().putStringProperty(Message.HDR_GROUP_ID, groupName);
+      queue.addTail(firstMessageReference, true);
+      Assert.assertTrue("first message isn't handled", firstMessageHandled.await(3000, TimeUnit.MILLISECONDS));
+      Assert.assertEquals("group consumer isn't correctly set", groupConsumer, queue.getGroups().get(groupName));
+      queue.addTail(secondMessageReference, true);
+      final boolean atLeastTwoDeliverAttempts = finished.await(3000, TimeUnit.MILLISECONDS);
+      Assert.assertTrue(atLeastTwoDeliverAttempts);
+      Thread.sleep(1000);
+      Assert.assertEquals("The second message should be in the queue", 1, queue.getMessageCount());
+   }
+
 }