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 2021/05/14 14:08:32 UTC
[activemq-artemis] branch main updated: ARTEMIS-3295 - do not
cluster advisories
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/main by this push:
new a34d9aa ARTEMIS-3295 - do not cluster advisories
a34d9aa is described below
commit a34d9aad6bb4f19678c12bddedda5a37509e2cf0
Author: Andy Taylor <an...@gmail.com>
AuthorDate: Thu May 13 12:08:33 2021 +0100
ARTEMIS-3295 - do not cluster advisories
https://issues.apache.org/jira/browse/ARTEMIS-3295
---
.../protocol/openwire/OpenWireProtocolManager.java | 3 ++
.../core/server/cluster/ClusterManager.java | 14 ++++++
.../cluster/impl/ClusterConnectionBridge.java | 6 +++
.../cluster/MessageRedistributionTest.java | 56 ++++++++++++++++++++++
4 files changed, 79 insertions(+)
diff --git a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManager.java b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManager.java
index 21e9748..3aa5868 100644
--- a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManager.java
+++ b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManager.java
@@ -177,6 +177,9 @@ public class OpenWireProtocolManager implements ProtocolManager<Interceptor>, Cl
if (cc != null) {
cc.addClusterTopologyListener(this);
}
+
+ //make sure we don't cluster advisories
+ clusterManager.addProtocolIgnoredAddress(AdvisorySupport.ADVISORY_TOPIC_PREFIX);
}
@Override
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ClusterManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ClusterManager.java
index 14d4af0..78bc50d 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ClusterManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/ClusterManager.java
@@ -101,6 +101,8 @@ public final class ClusterManager implements ActiveMQComponent {
private final Configuration configuration;
+ private Set<String> protocolIgnoredAddresses = new HashSet<>();
+
public QuorumManager getQuorumManager() {
return clusterController.getQuorumManager();
}
@@ -113,6 +115,7 @@ public final class ClusterManager implements ActiveMQComponent {
return haManager;
}
+
public void addClusterChannelHandler(Channel channel,
Acceptor acceptorUsed,
CoreRemotingConnection remotingConnection,
@@ -337,6 +340,8 @@ public final class ClusterManager implements ActiveMQComponent {
state = State.STOPPED;
clearClusterConnections();
+
+ protocolIgnoredAddresses.clear();
}
public void flushExecutor() {
@@ -574,6 +579,15 @@ public final class ClusterManager implements ActiveMQComponent {
}
}
+ public ClusterManager addProtocolIgnoredAddress(String ignoredAddress) {
+ protocolIgnoredAddresses.add(ignoredAddress);
+ return this;
+ }
+
+ public Collection<String> getProtocolIgnoredAddresses() {
+ return protocolIgnoredAddresses;
+ }
+
// Private methods ----------------------------------------------------------------------------------------------------
private void clearClusterConnections() {
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionBridge.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionBridge.java
index da161ac..ff9a179 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionBridge.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionBridge.java
@@ -75,6 +75,7 @@ public class ClusterConnectionBridge extends BridgeImpl {
private final SimpleString managementNotificationAddress;
+
private ClientConsumer notifConsumer;
private final SimpleString idsHeaderName;
@@ -86,6 +87,7 @@ public class ClusterConnectionBridge extends BridgeImpl {
private final ServerLocatorInternal discoveryLocator;
private final String storeAndForwardPrefix;
+
private TopologyMemberImpl member;
public ClusterConnectionBridge(final ClusterConnection clusterConnection,
@@ -362,6 +364,10 @@ public class ClusterConnectionBridge extends BridgeImpl {
}
filterString += "!" + storeAndForwardPrefix;
filterString += ",!" + managementAddress;
+ //add any protocol specific ignored addresses
+ for (String ignoreAddress : clusterManager.getProtocolIgnoredAddresses()) {
+ filterString += ",!" + ignoreAddress;
+ }
return filterString;
}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/cluster/MessageRedistributionTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/cluster/MessageRedistributionTest.java
index 083c13b..d83491a 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/cluster/MessageRedistributionTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/cluster/MessageRedistributionTest.java
@@ -28,11 +28,16 @@ import org.apache.activemq.artemis.tests.integration.cluster.distribution.Cluste
import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.util.ConsumerThread;
+import org.junit.Assert;
import org.junit.Test;
import javax.jms.Connection;
import javax.jms.Destination;
+import javax.jms.Message;
+import javax.jms.MessageProducer;
import javax.jms.Session;
+import javax.jms.Topic;
+import javax.jms.TopicSubscriber;
import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -71,6 +76,57 @@ public class MessageRedistributionTest extends ClusterTestBase {
}
}
+ @Test
+ public void testAdvisoriesNotClustered() throws Exception {
+
+ setupServer(0, true, true);
+ setupServer(1, true, true);
+
+ setupClusterConnection("cluster0", "", MessageLoadBalancingType.ON_DEMAND, 1, true, 0, 1);
+ setupClusterConnection("cluster1", "", MessageLoadBalancingType.ON_DEMAND, 1, true, 1, 0);
+
+ startServers(0, 1);
+
+ waitForTopology(servers[0], 2);
+ waitForTopology(servers[1], 2);
+
+ setupSessionFactory(0, true);
+ setupSessionFactory(1, true);
+
+ createAddressInfo(0, "testAddress", RoutingType.MULTICAST, -1, false);
+ createAddressInfo(1, "testAddress", RoutingType.MULTICAST, -1, false);
+
+ ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
+ ActiveMQConnectionFactory factory2 = new ActiveMQConnectionFactory(getServerUri(1));
+ Connection conn = null;
+ Connection conn2 = null;
+ CountDownLatch active = new CountDownLatch(1);
+ try {
+ conn = factory.createConnection();
+ conn2 = factory2.createConnection();
+ conn2.setClientID("id");
+ conn2.start();
+ Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Session session2 = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Topic dest = (Topic) ActiveMQDestination.createDestination("testAddress", ActiveMQDestination.TOPIC_TYPE);
+ TopicSubscriber mySubscriber = session2.createDurableSubscriber(dest, "mySubscriber");
+ MessageProducer producer = session.createProducer(dest);
+ producer.send(session.createTextMessage("test message"));
+ Message message = mySubscriber.receive(5000);
+ SimpleString advQueue = new SimpleString("ActiveMQ.Advisory.TempQueue");
+ SimpleString advTopic = new SimpleString("ActiveMQ.Advisory.TempTopic");
+ //we create a consumer on node 2 and assert that the advisory subscription queue is not clustered
+ Assert.assertEquals("", 1, servers[0].getPostOffice().getBindingsForAddress(advQueue).getBindings().size());
+ Assert.assertEquals("", 1, servers[0].getPostOffice().getBindingsForAddress(advTopic).getBindings().size());
+ Assert.assertEquals("", 1, servers[1].getPostOffice().getBindingsForAddress(advQueue).getBindings().size());
+ Assert.assertEquals("", 1, servers[1].getPostOffice().getBindingsForAddress(advTopic).getBindings().size());
+
+ } finally {
+ conn.close();
+ conn2.close();
+ }
+ }
+
@Override
protected boolean isResolveProtocols() {
return true;