Skip to content

ARTEMIS-6004 Add clientFailoverAdvertisingEnabled configuration to not send failover server list info to AMQP clients #6614

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
KalCramer wants to merge 1 commit into
base: main
Choose a base branch
from
Open

ARTEMIS-6004 Add clientFailoverAdvertisingEnabled configuration to not send failover server list info to AMQP clients #6614

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions ...rc/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,6 +150,8 @@ public class TransportConstants {

public static final String TRUST_MANAGER_FACTORY_PLUGIN_PROP_NAME = "trustManagerFactoryPlugin";

public static final String CLIENT_FAILOVER_ADVERTISING_ENABLED_PROP_NAME = "clientFailoverAdvertisingEnabled";

public static final String NETTY_VERSION;

/**
Expand Down
15 changes: 13 additions & 2 deletions ...rc/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallback.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,10 +27,12 @@
import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
import org.apache.activemq.artemis.core.client.impl.TopologyMemberImpl;
import org.apache.activemq.artemis.core.remoting.CloseListener;
import org.apache.activemq.artemis.core.remoting.FailureListener;
import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.cluster.ClusterConnection;
import org.apache.activemq.artemis.core.server.cluster.ClusterManager;
Expand All @@ -49,12 +51,14 @@
import org.apache.activemq.artemis.protocol.amqp.sasl.ServerSASLFactory;
import org.apache.activemq.artemis.spi.core.remoting.Connection;
import org.apache.activemq.artemis.spi.core.remoting.ReadyListener;
import org.apache.activemq.artemis.utils.ConfigurationHelper;
import org.apache.activemq.artemis.utils.UUIDGenerator;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.transport.AmqpError;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.invoke.MethodHandles;

import io.netty.buffer.ByteBuf;
Expand Down Expand Up @@ -191,7 +195,7 @@ public AMQPSessionCallback createSessionCallback(AMQPConnectionContext connectio
}

public void sendSASLSupported() {
connection.write(ActiveMQBuffers.wrappedBuffer(new byte[]{'A', 'M', 'Q', 'P', 3, 1, 0, 0}));
connection.write(ActiveMQBuffers.wrappedBuffer(new byte[] {'A', 'M', 'Q', 'P', 3, 1, 0, 0}));
}

public boolean validateConnection(org.apache.qpid.proton.engine.Connection connection, SASLResult saslResult) {
Expand Down Expand Up @@ -266,7 +270,14 @@ public URI getFailoverList() {
if (clusterConnection != null) {
TopologyMemberImpl member = clusterConnection.getTopology().getMember(server.getNodeID().toString());
if (member != null) {
return member.toBackupURI();
TransportConfiguration backupConnector = member.getBackup();

@tabish121 tabish121 Aug 18, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Adding this here seems off if you intend as you've said above for this to be a more generic option applicable to other protocols besides AMQP. Having to copy this code block around for any other implementation that needs it is sub optimal and likely error prone. Seems as though this code should live elsewhere so that it can be called here or in other protocols implementations to check if the connector is to be advertised.

@KalCramer KalCramer Aug 18, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Currently this fix is specific for AMQP. I could see making this naming AMQP specific if that make it more obvious for now. No matter what there won't be a single spot for all protocols unless an abstraction is specifically made for this but I don't think this is something all protocols support so I wouldn't think that makes sense but let me know if that would stop this getting merged. Core might be only other protocol that I know of that might use this but I could be wrong.

if (backupConnector == null) {
return null;
}
boolean clientFailoverAdvertisingEnabled = ConfigurationHelper.getBooleanProperty(TransportConstants.CLIENT_FAILOVER_ADVERTISING_ENABLED_PROP_NAME, true, backupConnector.getCombinedParams());
if (clientFailoverAdvertisingEnabled) {
return member.toBackupURI();
}
}
}
return null;
Expand Down
130 changes: 130 additions & 0 deletions ...est/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallbackTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,10 +23,18 @@
import static org.junit.jupiter.api.Assertions.assertTrue;

import org.apache.activemq.artemis.api.core.ActiveMQSecurityException;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.core.client.impl.Topology;
import org.apache.activemq.artemis.core.client.impl.TopologyMemberImpl;
import org.apache.activemq.artemis.core.remoting.impl.invm.InVMConnection;
import org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnection;
import org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnectorFactory;
import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants;
import org.apache.activemq.artemis.core.security.SecurityStore;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.cluster.ClusterConnection;
import org.apache.activemq.artemis.core.server.cluster.ClusterManager;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
import org.apache.activemq.artemis.protocol.amqp.sasl.AnonymousServerSASL;
import org.apache.activemq.artemis.protocol.amqp.sasl.GSSAPIServerSASL;
Expand All @@ -36,6 +44,10 @@
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;

import java.net.URI;
import java.util.HashMap;
import java.util.Map;

public class AMQPConnectionCallbackTest {

@Test
Expand Down Expand Up @@ -93,4 +105,122 @@ public void testAnonymousSupportCheck() throws Exception {
assertFalse(callback.isSupportsAnonymous());
Mockito.verify(securityStore).authenticate(Mockito.any(), Mockito.any(), Mockito.same(connectionDelegate));
}

@Test
public void testGetFailoverListFailoverEnabled() throws Exception {
ActiveMQServer server = Mockito.mock(ActiveMQServer.class);

TransportConfiguration backup = createBackupTransportConfiguration();
backup.getParams().put(TransportConstants.CLIENT_FAILOVER_ADVERTISING_ENABLED_PROP_NAME, true);

AMQPConnectionCallback callback = createCallback(server, backup);

URI failoverList = callback.getFailoverList();

assertNotNull(failoverList);
assertEquals("tcp", failoverList.getScheme());
assertEquals("backup", failoverList.getHost());
assertEquals(61617, failoverList.getPort());
assertEquals("sslEnabled=false", failoverList.getQuery());
}

@Test
public void testGetFailoverListFailoverDisabled() throws Exception {
ActiveMQServer server = Mockito.mock(ActiveMQServer.class);

TransportConfiguration backup = createBackupTransportConfiguration();
backup.getParams().put(TransportConstants.CLIENT_FAILOVER_ADVERTISING_ENABLED_PROP_NAME, false);

AMQPConnectionCallback callback = createCallback(server, backup);

assertNull(callback.getFailoverList());
}

@Test
public void testGetFailoverListFailoverEnabledByDefault() throws Exception {
ActiveMQServer server = Mockito.mock(ActiveMQServer.class);

TransportConfiguration backup = createBackupTransportConfiguration();

// CLIENT_FAILOVER_ADVERTISING_ENABLED_PROP_NAME is intentionally not set.
// The default value is true.
AMQPConnectionCallback callback = createCallback(server, backup);

URI failoverList = callback.getFailoverList();

assertNotNull(failoverList);
assertEquals("tcp", failoverList.getScheme());
assertEquals("backup", failoverList.getHost());
assertEquals(61617, failoverList.getPort());
assertEquals("sslEnabled=false", failoverList.getQuery());
}

@Test
Comment thread
KalCramer marked this conversation as resolved.
public void testGetFailoverListIncludesSslEnabled() throws Exception {
ActiveMQServer server = Mockito.mock(ActiveMQServer.class);

TransportConfiguration backup = createBackupTransportConfiguration();
backup.getParams().put(
TransportConstants.SSL_ENABLED_PROP_NAME,
true);
backup.getParams().put(
TransportConstants.CLIENT_FAILOVER_ADVERTISING_ENABLED_PROP_NAME,
true);

AMQPConnectionCallback callback = createCallback(server, backup);

URI failoverList = callback.getFailoverList();

assertNotNull(failoverList);
assertEquals("tcp", failoverList.getScheme());
assertEquals("backup", failoverList.getHost());
assertEquals(61617, failoverList.getPort());
assertEquals("sslEnabled=true", failoverList.getQuery());
}

private TransportConfiguration createBackupTransportConfiguration() {
Map<String, Object> params = new HashMap<>();
params.put(TransportConstants.HOST_PROP_NAME, "backup");
params.put(TransportConstants.PORT_PROP_NAME, 61617);

return new TransportConfiguration(
NettyConnectorFactory.class.getName(),
params);
}


private AMQPConnectionCallback createCallback(
ActiveMQServer server,
TransportConfiguration backup) throws Exception {

String nodeId = "test-node";

Mockito.when(server.getNodeID()).thenReturn(SimpleString.of(nodeId));

ClusterManager clusterManager = Mockito.mock(ClusterManager.class);
ClusterConnection clusterConnection = Mockito.mock(ClusterConnection.class);
Topology topology = Mockito.mock(Topology.class);

Mockito.when(server.getClusterManager()).thenReturn(clusterManager);
Mockito.when(clusterManager.getDefaultConnection(null)).thenReturn(clusterConnection);
Mockito.when(clusterConnection.getTopology()).thenReturn(topology);

TopologyMemberImpl member = new TopologyMemberImpl(
nodeId,
null,
null,
null,
backup);

Mockito.when(topology.getMember(nodeId)).thenReturn(member);

ProtonProtocolManager protocolManager = Mockito.mock(ProtonProtocolManager.class);
Mockito.when(protocolManager.getServer()).thenReturn(server);

return new AMQPConnectionCallback(
protocolManager,
null,
null,
server);
}
}
5 changes: 5 additions & 0 deletions docs/user-manual/configuring-transports.adoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -491,6 +491,11 @@ This value takes precedence of all other SSL parameters which apply to the trust
+
Any plugin specified will need to be placed on the xref:using-server.adoc#adding-runtime-dependencies[broker's classpath].

clientFailoverAdvertisingEnabled::
When used on a `connector` determines whether the connector is advertised to clients as a failover connector.
Valid values are `true` or `false`.
Default is `true`.

==== Configuring an SSLContextFactory

If you use `JDK` as SSL provider (the default), you can configure which SSLContextFactory to use.
Expand Down