You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@artemis.apache.org by cl...@apache.org on 2026/09/10 17:09:20 UTC
(artemis) 07/11: Fix compatibility with older clients during pre-authentication topology exchange.
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/artemis.git
commit ffb79bb98704284ed8cc2ab0cec1f35637c4c858
Author: Clebert Suconic <cl...@gmail.com>
AuthorDate: Fri Aug 7 13:07:37 2026 -0400
Fix compatibility with older clients during pre-authentication topology exchange.
Before authentication, the server sends a placeholder topology entry so clients don't
block waiting for topology data. Previously, this placeholder used null connectors with
a dummy name (PRE_AUTH_CONNECTOR), which older clients couldn't handle.
The fix now sends the actual connector for the node the client is already connected to.
Once the client authenticates, the placeholder is replaced with the real topology.
(cherry picked from commit 96eae4cee2de363af8dc381a5c28772437af050c)
ARTEMIS-6246
---
.../core/impl/BackwardsCompatibilityUtils.java | 8 ++
.../protocol/core/impl/CoreProtocolManager.java | 22 +++++-
.../legacyclient/artemisTopologyTest.groovy | 89 +++++++++++++++-------
.../resources/legacyclient/hqTopologyTest.groovy | 78 ++++++++++++-------
.../compatibility/LegacyClientTopologyTest.java | 46 +++++++++--
5 files changed, 178 insertions(+), 65 deletions(-)
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/BackwardsCompatibilityUtils.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/BackwardsCompatibilityUtils.java
index 2bcdf13e77..0fbc7d3c14 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/BackwardsCompatibilityUtils.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/BackwardsCompatibilityUtils.java
@@ -168,6 +168,14 @@ public class BackwardsCompatibilityUtils {
return new Pair<>(member.getPrimary(), member.getBackup());
}
+ public static Pair<TransportConfiguration, TransportConfiguration> checkTCPPairConversion(int clientIncrementingVersion,
+ Pair<TransportConfiguration, TransportConfiguration> pair) {
+ if (clientIncrementingVersion < INITIAL_ACTIVEMQ_INCREMENTING_VERSION) {
+ return new Pair<>(convertTransport(pair.getA()), convertTransport(pair.getB()));
+ }
+ return pair;
+ }
+
/**
* Replaces class name and parameter names to HornetQ values.
*/
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/CoreProtocolManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/CoreProtocolManager.java
index ce8cad8d2f..807cc3cdcf 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/CoreProtocolManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/CoreProtocolManager.java
@@ -381,12 +381,28 @@ public class CoreProtocolManager implements ProtocolManager<Interceptor, ActiveM
// Send a single dummy topology response to unblock the client's isLast=true wait (the client blocks
// until it receives a topology message before proceeding to CREATE_SESSION). The dummy response
- // uses PRE_AUTH_NODE_ID and an empty connector pair so no real broker identity or connectivity
+ // uses PRE_AUTH_NODE_ID. The current connector is sent so older clients wouldn't fail and would have something they can connect to.
// information is disclosed to unauthenticated callers.
+ // Notice that this PRE_AUTH node is removed once the node is connected
private void sendPreAuthTopology() {
preAuthTopologySent = true;
- Pair<TransportConfiguration, TransportConfiguration> preAuthConfig = new Pair<>(null, new TransportConfiguration(
- rc.getTransportConnection().getConnectorConfig().getFactoryClassName(), null, Topology.PRE_AUTH_CONNECTOR_NAME));
+ TransportConfiguration connectorConfig = rc.getTransportConnection().getConnectorConfig();
+ Map<String, Object> connectorParams = new HashMap<>();
+ if (connectorConfig != null && connectorConfig.getParams() != null) {
+ Object host = connectorConfig.getParams().get(org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.HOST_PROP_NAME);
+ Object port = connectorConfig.getParams().get(org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.PORT_PROP_NAME);
+ if (host != null) {
+ connectorParams.put(org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.HOST_PROP_NAME, host);
+ }
+ if (port != null) {
+ connectorParams.put(org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.PORT_PROP_NAME, port);
+ }
+ }
+ String factoryClassName = connectorConfig != null ? connectorConfig.getFactoryClassName() : org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnectorFactory.class.getName();
+ TransportConfiguration liveConnector = new TransportConfiguration(factoryClassName, connectorParams);
+ Pair<TransportConfiguration, TransportConfiguration> preAuthConfig = BackwardsCompatibilityUtils.checkTCPPairConversion(
+ channel0.getConnection().getChannelVersion(),
+ new Pair<>(liveConnector, new TransportConfiguration(factoryClassName, null, Topology.PRE_AUTH_CONNECTOR_NAME)));
logger.debug("Sending pre-authentication topology to unauthenticated connection from {}.", rc.getRemoteAddress());
diff --git a/tests/compatibility-tests/src/main/resources/legacyclient/artemisTopologyTest.groovy b/tests/compatibility-tests/src/main/resources/legacyclient/artemisTopologyTest.groovy
index 1525b0d628..cd746d9fc0 100644
--- a/tests/compatibility-tests/src/main/resources/legacyclient/artemisTopologyTest.groovy
+++ b/tests/compatibility-tests/src/main/resources/legacyclient/artemisTopologyTest.groovy
@@ -19,27 +19,48 @@ package legacyclient
import org.apache.activemq.artemis.api.core.TransportConfiguration
import org.apache.activemq.artemis.api.core.client.ActiveMQClient
+import org.apache.activemq.artemis.api.core.client.ClientSession
import org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnectorFactory
import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants
import java.lang.reflect.Field
-def printTopology(locator, String label) {
+def getLiveConnector(member) {
try {
- Field topologyField = locator.getClass().getDeclaredField("topologyArray")
- topologyField.setAccessible(true)
- def topologyArray = topologyField.get(locator)
- if (topologyArray != null) {
- println(label + " topologyArray length: " + topologyArray.length)
- for (int t = 0; t < topologyArray.length; t++) {
- def pair = topologyArray[t]
- println(" [" + t + "] A=" + pair.getA() + " B=" + pair.getB())
+ return member.getPrimary()
+ } catch (Exception e) {
+ return member.getLive()
+ }
+}
+
+def printTopology(locator, String label, boolean failIfPreAuth) {
+ String preAuthError = null
+ try {
+ def topology = locator.getTopology()
+ def members = topology.getMembers()
+ println(label + " topology members: " + members.size())
+ for (def member : members) {
+ def live = getLiveConnector(member)
+ println(" nodeID=" + member.getNodeId() + " live=" + live + " backup=" + member.getBackup())
+ if (failIfPreAuth) {
+ if (live != null && "PRE_AUTH_CONNECTOR".equals(live.getName())) {
+ preAuthError = "PRE_AUTH_CONNECTOR found in live for nodeID=" + member.getNodeId()
+ }
+ if (member.getBackup() != null && "PRE_AUTH_CONNECTOR".equals(member.getBackup().getName())) {
+ preAuthError = "PRE_AUTH_CONNECTOR found in backup for nodeID=" + member.getNodeId()
+ }
}
- } else {
- println(label + " topologyArray is null")
}
+ if (preAuthError != null) {
+ throw new Exception(preAuthError)
+ }
+ return members.size()
} catch (Exception e) {
+ if (preAuthError != null) {
+ throw e
+ }
println(label + " Could not inspect topology: " + e.getMessage())
+ throw e
}
}
@@ -50,29 +71,39 @@ def tc = new TransportConfiguration(NettyConnectorFactory.class.getName(), param
def locator = ActiveMQClient.createServerLocatorWithHA(tc)
-printTopology(locator, "Before first createSessionFactory:")
+printTopology(locator, "Before first createSessionFactory:", false)
println("=== Attempting first createSessionFactory ===")
-try {
- def sf = locator.createSessionFactory()
- println("first createSessionFactory succeeded")
+def sf = locator.createSessionFactory()
+println("first createSessionFactory succeeded")
- printTopology(locator, "After first createSessionFactory:")
- println("=== Attempting second createSessionFactory ===")
- def sf2 = locator.createSessionFactory()
- println("second createSessionFactory succeeded")
- printTopology(locator, "After second createSessionFactory:")
+def count = printTopology(locator, "After first createSessionFactory:", false)
+if (count > 1) {
+ throw new Exception("Topology after first createSessionFactory has " + count1 + " elements, expected at most 1")
+}
+
+println("=== Attempting second createSessionFactory ===")
+def sf2 = locator.createSessionFactory()
+println("second createSessionFactory succeeded")
+
+count = printTopology(locator, "After second createSessionFactory:", false)
+if (count > 1) {
+ throw new Exception("Topology after second createSessionFactory has " + count2 + " elements, expected at most 1")
+}
+
+ClientSession session = sf.createSession("guest", "guest", false, true, true, false, 0)
+
+Thread.sleep(1000);
- sf2.close()
- sf.close()
- locator.close()
- return true
-} catch (Exception e) {
- println("Exception: " + e.getClass().getName() + " - " + e.getMessage())
- e.printStackTrace()
- locator.close()
- return false
+count = printTopology(locator, "After authorization:", true)
+if (count != 2) {
+ throw new Exception("Cluster topology is not correctly informed to the client")
}
+session.close();
+sf2.close()
+sf.close()
+locator.close()
+return true
diff --git a/tests/compatibility-tests/src/main/resources/legacyclient/hqTopologyTest.groovy b/tests/compatibility-tests/src/main/resources/legacyclient/hqTopologyTest.groovy
index 5af8159bf9..52d8d6efd1 100644
--- a/tests/compatibility-tests/src/main/resources/legacyclient/hqTopologyTest.groovy
+++ b/tests/compatibility-tests/src/main/resources/legacyclient/hqTopologyTest.groovy
@@ -18,28 +18,40 @@ package legacyclient
*/
import org.hornetq.api.core.TransportConfiguration
+import org.hornetq.api.core.client.ClientSession
import org.hornetq.api.core.client.HornetQClient
import org.hornetq.core.remoting.impl.netty.NettyConnectorFactory
import org.hornetq.core.remoting.impl.netty.TransportConstants
import java.lang.reflect.Field
-def printTopology(locator, String label) {
+def printTopology(locator, String label, boolean failIfPreAuth) {
+ String preAuthError = null
try {
- Field topologyField = locator.getClass().getDeclaredField("topologyArray")
- topologyField.setAccessible(true)
- def topologyArray = topologyField.get(locator)
- if (topologyArray != null) {
- println(label + " topologyArray length: " + topologyArray.length)
- for (int t = 0; t < topologyArray.length; t++) {
- def pair = topologyArray[t]
- println(" [" + t + "] A=" + pair.getA() + " B=" + pair.getB())
+ def topology = locator.getTopology()
+ def members = topology.getMembers()
+ println(label + " topology members: " + members.size())
+ for (def member : members) {
+ println(" nodeID=" + member.getNodeId() + " live=" + member.getLive() + " backup=" + member.getBackup())
+ if (failIfPreAuth) {
+ if (member.getLive() != null && "PRE_AUTH_CONNECTOR".equals(member.getLive().getName())) {
+ preAuthError = "PRE_AUTH_CONNECTOR found in live for nodeID=" + member.getNodeId()
+ }
+ if (member.getBackup() != null && "PRE_AUTH_CONNECTOR".equals(member.getBackup().getName())) {
+ preAuthError = "PRE_AUTH_CONNECTOR found in backup for nodeID=" + member.getNodeId()
+ }
}
- } else {
- println(label + " topologyArray is null")
}
+ if (preAuthError != null) {
+ throw new Exception(preAuthError)
+ }
+ return members.size()
} catch (Exception e) {
+ if (preAuthError != null) {
+ throw e
+ }
println(label + " Could not inspect topology: " + e.getMessage())
+ return 0
}
}
@@ -50,29 +62,39 @@ def tc = new TransportConfiguration(NettyConnectorFactory.class.getName(), param
def locator = HornetQClient.createServerLocatorWithHA(tc)
-printTopology(locator, "Before first createSessionFactory:")
+printTopology(locator, "Before first createSessionFactory:", false)
println("=== Attempting createSessionFactory (subscribes to topology before auth) ===")
-try {
- def sf = locator.createSessionFactory()
- println("createSessionFactory succeeded")
+def sf = locator.createSessionFactory()
+println("createSessionFactory succeeded")
+
+def count = printTopology(locator, "After first createSessionFactory:", false)
+if (count > 1) {
+ throw new Exception("Topology after first createSessionFactory has " + count1 + " elements, expected at most 1")
+}
+
+println("=== Attempting second createSessionFactory (should use topologyArray) ===")
+def sf2 = locator.createSessionFactory()
+println("second createSessionFactory succeeded")
- printTopology(locator, "After first createSessionFactory:")
+count = printTopology(locator, "After second createSessionFactory:", false)
+if (count > 1) {
+ throw new Exception("Topology after second createSessionFactory has " + count2 + " elements, expected at most 1")
+}
- println("=== Attempting second createSessionFactory (should use topologyArray) ===")
- def sf2 = locator.createSessionFactory()
- println("second createSessionFactory succeeded")
+ClientSession session = sf.createSession("guest", "guest", false, true, true, false, 0)
- printTopology(locator, "After second createSessionFactory:")
- sf2.close()
+Thread.sleep(1000);
- locator.close()
- return true
-} catch (Exception e) {
- println("Exception: " + e.getClass().getName() + " - " + e.getMessage())
- e.printStackTrace()
- locator.close()
- return false
+count = printTopology(locator, "After authorization:", true)
+if (count != 2) {
+ throw new Exception("Cluster topology is not correctly informed to the client")
}
+session.close();
+System.out.println("count : " + count)
+sf2.close()
+sf.close()
+locator.close()
+return true
diff --git a/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/LegacyClientTopologyTest.java b/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/LegacyClientTopologyTest.java
index ec6701f390..aec7083d80 100644
--- a/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/LegacyClientTopologyTest.java
+++ b/tests/compatibility-tests/src/test/java/org/apache/activemq/artemis/tests/compatibility/LegacyClientTopologyTest.java
@@ -30,6 +30,7 @@ import static org.apache.activemq.artemis.tests.compatibility.GroovyRun.HORNETQ_
import static org.apache.activemq.artemis.tests.compatibility.GroovyRun.SNAPSHOT;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
@@ -43,6 +44,7 @@ import org.apache.activemq.artemis.core.server.ActiveMQServers;
import org.apache.activemq.artemis.core.server.JournalType;
import org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager;
import org.apache.activemq.artemis.tests.compatibility.base.ClasspathBase;
+import org.apache.activemq.artemis.utils.Wait;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -50,6 +52,7 @@ import org.junit.jupiter.api.Test;
public class LegacyClientTopologyTest extends ClasspathBase {
private ActiveMQServer server;
+ private ActiveMQServer server2;
@BeforeEach
public void setUp() throws Exception {
@@ -58,25 +61,58 @@ public class LegacyClientTopologyTest extends ClasspathBase {
configuration.addAcceptorConfiguration("artemis", "tcp://0.0.0.0:61616?anycastPrefix=jms.queue.&multicastPrefix=jms.topic.");
configuration.setSecurityEnabled(true);
configuration.setPersistenceEnabled(false);
+ configuration.setClusterPassword("guest");
configuration.addConnectorConfiguration("netty-connector", "tcp://localhost:61616");
- configuration.addClusterConfiguration(new ClusterConnectionConfiguration().setName("my-cluster").setConnectorName("netty-connector"));
+ configuration.addConnectorConfiguration("netty-connector-2", "tcp://localhost:61617");
+ configuration.addClusterConfiguration(new ClusterConnectionConfiguration().setName("my-cluster").setConnectorName("netty-connector").setStaticConnectors(Arrays.asList("netty-connector-2")));
server = ActiveMQServers.newActiveMQServer(configuration, false);
server.start();
- ActiveMQJAASSecurityManager securityManager = (ActiveMQJAASSecurityManager) server.getSecurityManager();
+ ConfigurationImpl configuration2 = new ConfigurationImpl();
+ configuration2.setJournalType(JournalType.NIO);
+ configuration2.addAcceptorConfiguration("artemis", "tcp://0.0.0.0:61617?anycastPrefix=jms.queue.&multicastPrefix=jms.topic.");
+ configuration2.setSecurityEnabled(true);
+ configuration2.setPersistenceEnabled(false);
+ configuration2.setClusterPassword("guest");
+ configuration2.addConnectorConfiguration("netty-connector", "tcp://localhost:61616");
+ configuration2.addConnectorConfiguration("netty-connector-2", "tcp://localhost:61617");
+ configuration2.addClusterConfiguration(new ClusterConnectionConfiguration().setName("my-cluster").setConnectorName("netty-connector-2").setStaticConnectors(Arrays.asList("netty-connector")));
+
+ server2 = ActiveMQServers.newActiveMQServer(configuration2, false);
+ server2.start();
+
+ setupSecurity(server);
+ setupSecurity(server2);
+
+
+ Wait.assertTrue(() -> {
+ for (org.apache.activemq.artemis.core.server.cluster.ClusterConnection cc : server.getClusterManager().getClusterConnections()) {
+ if (cc.getTopology().getMembers().size() == 2) {
+ return true;
+ }
+ }
+ return false;
+ }, 10000, 100);
+
+ server.createQueue(QueueConfiguration.of("jms.queue.testQueue").setRoutingType(RoutingType.ANYCAST).setDurable(false));
+ }
+
+ private void setupSecurity(ActiveMQServer srv) {
+ ActiveMQJAASSecurityManager securityManager = (ActiveMQJAASSecurityManager) srv.getSecurityManager();
securityManager.getConfiguration().addUser("guest", "guest");
securityManager.getConfiguration().addRole("guest", "guest");
Set<Role> roles = new HashSet<>();
roles.add(new Role("guest", true, true, true, true, true, true, true, true, true, true, false, false));
- server.getSecurityRepository().addMatch("#", roles);
-
- server.createQueue(QueueConfiguration.of("jms.queue.testQueue").setRoutingType(RoutingType.ANYCAST).setDurable(false));
+ srv.getSecurityRepository().addMatch("#", roles);
}
@AfterEach
public void tearDown() throws Exception {
+ if (server2 != null) {
+ server2.stop();
+ }
if (server != null) {
server.stop();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@artemis.apache.org
For additional commands, e-mail: commits-help@artemis.apache.org