From 8d848f4615a6ea81002eaf638e93826c810b0884 Mon Sep 17 00:00:00 2001
From: gtully <gary.tully@gmail.com>
Date: Thu, 18 Mar 2021 10:08:38 +0000
Subject: [PATCH] ARTEMIS-3188 / ARTEMIS-2108 Fixing StackOverFlow over
 bindings

This fix was done in collaboration with Gary Tully
---
 .../core/postoffice/impl/BindingsImpl.java    | 32 +++++++------------
 ...RemoteBindingWithoutLoadBalancingTest.java | 21 ++++++++++++
 2 files changed, 32 insertions(+), 21 deletions(-)

diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/BindingsImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/BindingsImpl.java
index bc29edb51d..54266e15fe 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/BindingsImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/BindingsImpl.java
@@ -437,9 +437,7 @@ public final class BindingsImpl implements Bindings {
             }
          }
 
-         Filter filter = binding.getFilter();
-
-         if (filter == null || filter.match(message)) {
+         if (matchBinding(message, binding)) {
             // bindings.length == 1 ==> only a local queue so we don't check for matching consumers (it's an
             // unnecessary overhead)
             if (length == 1 || (binding.isConnected() && (messageLoadBalancingType.equals(MessageLoadBalancingType.STRICT) || binding.isHighAcceptPriority(message)))) {
@@ -486,29 +484,21 @@ public final class BindingsImpl implements Bindings {
       if (pos != startPos) {
          routingNamePositions.put(routingName, pos);
       }
-
-      if (messageLoadBalancingType.equals(MessageLoadBalancingType.OFF) && theBinding instanceof RemoteQueueBinding) {
-         if (exclusivelyRemote(bindings)) {
-            theBinding = null;
-         } else {
-            theBinding = getNextBinding(message, routingName, bindings);
-         }
-      }
-
       return theBinding;
    }
 
-   private boolean exclusivelyRemote(List<Binding> bindings) {
-      boolean result = true;
-
-      for (Binding binding : bindings) {
-         if (!(binding instanceof RemoteQueueBinding)) {
-            result = false;
-            break;
-         }
+   private boolean matchBinding(Message message, Binding binding) {
+      if (messageLoadBalancingType.equals(MessageLoadBalancingType.OFF) && binding instanceof RemoteQueueBinding) {
+         return false;
       }
 
-      return result;
+      Filter filter = binding.getFilter();
+
+      if (filter == null || filter.match(message)) {
+         return true;
+      } else {
+         return false;
+      }
    }
 
    private void routeUsingStrictOrdering(final Message message,
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/RemoteBindingWithoutLoadBalancingTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/RemoteBindingWithoutLoadBalancingTest.java
index b627e8702f..028bbd7dcc 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/RemoteBindingWithoutLoadBalancingTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/distribution/RemoteBindingWithoutLoadBalancingTest.java
@@ -66,6 +66,27 @@ public class RemoteBindingWithoutLoadBalancingTest extends ClusterTestBase {
       send(1, "queues.testaddress", 1, false, null);
    }
 
+   @Test
+   public void testStackOverflowWithLocalConsumerAndFilter() throws Exception {
+      setupCluster();
+
+      startServers();
+
+      setupSessionFactory(0, isNetty());
+      setupSessionFactory(1, isNetty());
+
+      createQueue(0, "queues.testaddress", "queue0", "0", true);
+      createQueue(1, "queues.testaddress", "queue0", "1", true);
+
+      waitForBindings(0, "queues.testaddress", 1, 0, true);
+
+      waitForBindings(1, "queues.testaddress", 1, 0, false);
+
+      for (int i = 0; i < 10; i++) {
+         send(1, "queues.testaddress", 10, false, "" + i % 2);
+      }
+   }
+
    @Test
    public void testStackOverflowJMS() throws Exception {
       final String QUEUE_NAME = "queues.queue0";