You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2020/09/16 14:08:00 UTC
[activemq-artemis] branch master updated: ARTEMIS-2854 Non-durable
subscribers stop receiving after failover
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/master by this push:
new fe5b81f ARTEMIS-2854 Non-durable subscribers stop receiving after failover
new 42d7c33 This closes #3228
fe5b81f is described below
commit fe5b81fd5564bcc1a6c621cee776285297247964
Author: Howard Gao <ho...@gmail.com>
AuthorDate: Mon Aug 3 19:47:12 2020 +0800
ARTEMIS-2854 Non-durable subscribers stop receiving after failover
In a cluster scenario where non durable subscribers fail over to
backup while another live node forwarding messages to it,
there is a chance that the the live node keeps the old remote
binding for the subs and messages go to those
old remote bindings will result in "binding not found".
---
.../server/cluster/impl/ClusterConnectionImpl.java | 26 ++++--
.../bridge/ClusteredBridgeReconnectTest.java | 100 +++++++++++++++++++++
2 files changed, 117 insertions(+), 9 deletions(-)
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionImpl.java
index 03c5c0c..f952ef2 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionImpl.java
@@ -1283,17 +1283,25 @@ public final class ClusterConnectionImpl implements ClusterConnection, AfterConn
RemoteQueueBinding existingBinding = (RemoteQueueBinding) postOffice.getBinding(clusterName);
if (existingBinding != null) {
- if (!existingBinding.isConnected()) {
- existingBinding.connect();
+ if (queueID.equals(existingBinding.getRemoteQueueID())) {
+ if (!existingBinding.isConnected()) {
+ existingBinding.connect();
+ return;
+ }
+ // Sanity check - this means the binding has already been added via another bridge, probably max
+ // hops is too high
+ // or there are multiple cluster connections for the same address
+
+ ActiveMQServerLogger.LOGGER.remoteQueueAlreadyBoundOnClusterConnection(this, clusterName);
return;
}
- // Sanity check - this means the binding has already been added via another bridge, probably max
- // hops is too high
- // or there are multiple cluster connections for the same address
-
- ActiveMQServerLogger.LOGGER.remoteQueueAlreadyBoundOnClusterConnection(this, clusterName);
-
- return;
+ //this could happen during jms non-durable failover while the qname doesn't change but qid
+ //will be re-generated in backup. In that case a new remote binding will be created
+ //and put it to the map and old binding removed.
+ if (logger.isTraceEnabled()) {
+ logger.trace("Removing binding because qid changed " + queueID + " old: " + existingBinding.getRemoteQueueID());
+ }
+ removeBinding(clusterName);
}
RemoteQueueBinding binding = new RemoteQueueBindingImpl(server.getStorageManager().generateID(), queueAddress, clusterName, routingName, queueID, filterString, queue, bridge.getName(), distance + 1, messageLoadBalancingType);
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/bridge/ClusteredBridgeReconnectTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/bridge/ClusteredBridgeReconnectTest.java
index 060233c..aef4534 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/bridge/ClusteredBridgeReconnectTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/bridge/ClusteredBridgeReconnectTest.java
@@ -17,8 +17,10 @@
package org.apache.activemq.artemis.tests.integration.cluster.bridge;
import java.util.ArrayList;
+import java.util.UUID;
import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.client.ClientMessage;
import org.apache.activemq.artemis.api.core.client.ClientProducer;
import org.apache.activemq.artemis.api.core.client.ClientSession;
@@ -156,6 +158,104 @@ public class ClusteredBridgeReconnectTest extends ClusterTestBase {
}
+ @Test
+ public void testClusterBridgeAddRemoteBinding() throws Exception {
+
+ final String ADDRESS = "queues.testaddress";
+ final String QUEUE = UUID.randomUUID().toString();
+
+ setupServer(0, isFileStorage(), isNetty());
+ setupServer(1, isFileStorage(), isNetty());
+
+ setupClusterConnection("cluster0", "queues", MessageLoadBalancingType.ON_DEMAND, 1, isNetty(), 0, 1);
+
+ setupClusterConnection("cluster1", "queues", MessageLoadBalancingType.ON_DEMAND, 1, isNetty(), 1, 0);
+
+ startServers(0, 1);
+
+ setupSessionFactory(0, isNetty());
+ setupSessionFactory(1, isNetty());
+
+ createQueue(0, ADDRESS, QUEUE, null, false);
+
+ addConsumer(0, 0, QUEUE, null);
+
+ waitForBindings(0, ADDRESS, 1, 1, true);
+ waitForBindings(1, ADDRESS, 0, 0, true);
+
+ waitForBindings(0, ADDRESS, 0, 0, false);
+ waitForBindings(1, ADDRESS, 1, 1, false);
+
+ ClientSession session0 = sfs[0].createSession();
+ ClientSession session1 = sfs[1].createSession();
+
+ session0.start();
+ session1.start();
+
+ ClientProducer producer1 = session1.createProducer(ADDRESS);
+
+ int NUMBER_OF_MESSAGES = 10;
+
+ //send to node1 and receive from node0
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
+ ClientMessage msg = session1.createMessage(true);
+ producer1.send(msg);
+ session1.commit();
+ }
+
+ int cons0Count = 0;
+
+ while (true) {
+ ClientMessage msg = consumers[0].getConsumer().receive(1000);
+ if (msg == null) {
+ break;
+ }
+ cons0Count++;
+ msg.acknowledge();
+ session0.commit();
+ }
+ assertEquals(NUMBER_OF_MESSAGES, cons0Count);
+
+ //The following code similuates issue where a jms non-subscriber
+ //fails over to backup. In the process the temp queue is recreated
+ //on the backup with a new id while it's remote binding
+ //is created on the other node.
+ removeConsumer(0);
+ servers[0].getManagementService().enableNotifications(false);
+ servers[0].destroyQueue(new SimpleString(QUEUE));
+ servers[0].getManagementService().enableNotifications(true);
+
+ createQueue(0, ADDRESS, QUEUE, null, false);
+
+ addConsumer(0, 0, QUEUE, null);
+
+ waitForBindings(0, ADDRESS, 1, 1, true);
+ waitForBindings(1, ADDRESS, 0, 0, true);
+
+ waitForBindings(0, ADDRESS, 0, 0, false);
+ waitForBindings(1, ADDRESS, 1, 1, false);
+
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
+ ClientMessage msg = session1.createMessage(true);
+ producer1.send(msg);
+ session1.commit();
+ }
+
+ cons0Count = 0;
+
+ while (true) {
+ ClientMessage msg = consumers[0].getConsumer().receive(2000);
+ if (msg == null) {
+ break;
+ }
+ cons0Count++;
+ msg.acknowledge();
+ session0.commit();
+ }
+ assertEquals(NUMBER_OF_MESSAGES, cons0Count);
+
+ stopServers(0, 1);
+ }
@Override