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

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java

@@ -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;
 
    /**

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallback.java

@@ -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;
@@ -49,6 +51,7 @@
 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;
@@ -266,7 +269,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();
+            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;

artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallbackTest.java

@@ -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;
@@ -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
@@ -93,4 +105,121 @@ 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
+   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);
+   }
 }

artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPConnectionContextTest.java

@@ -142,4 +142,59 @@ public void testMaxTransferPerDeliveryAppliedToConnectionContext() throws Except
       connectionContext.onRemoteOpen(connection);
       connectionContext.close(null);
    }
+
+
+   @Test
+   public void testFailoverListNoFailover() throws Exception {
+      ArtemisExecutor executor = Mockito.mock(ArtemisExecutor.class);
+      ExecutorFactory executorFactory = Mockito.mock(ExecutorFactory.class);
+      Mockito.when(executorFactory.getExecutor()).thenReturn(executor);
+
+      ActiveMQServer server = Mockito.mock(ActiveMQServer.class);
+      Mockito.when(server.getExecutorFactory()).thenReturn(executorFactory);
+
+      final int expectedMaxTransfersPerDelivery = AmqpSupport.DEFAULT_MAX_TRANSFERS_PER_DELIVERY + 10;
+
+      ProtonProtocolManager manager = Mockito.mock(ProtonProtocolManager.class);
+      Mockito.when(manager.getServer()).thenReturn(server);
+      Mockito.when(manager.getMaxFrameSize()).thenReturn(AmqpSupport.MAX_FRAME_SIZE_DEFAULT);
+      Mockito.when(manager.getInitialRemoteMaxFrameSize()).thenReturn(AmqpSupport.INITIAL_REMOTE_MAX_FRAME_SIZE_DEFAULT);
+      Mockito.when(manager.getMaxTransfersPerDelivery()).thenReturn(expectedMaxTransfersPerDelivery);
+
+      EventLoop eventLoop = Mockito.mock(EventLoop.class);
+      Channel transportChannel = Mockito.mock(Channel.class);
+      Mockito.when(transportChannel.config()).thenReturn(Mockito.mock(ChannelConfig.class));
+      Mockito.when(transportChannel.eventLoop()).thenReturn(eventLoop);
+      Mockito.when(eventLoop.inEventLoop()).thenReturn(true);
+      NettyConnection transportConnection = new NettyConnection(new HashMap<>(), transportChannel, null, false, false);
+
+      Connection connection = Mockito.mock(Connection.class);
+      AMQPConnectionCallback protonSPI = Mockito.mock(AMQPConnectionCallback.class);
+      Mockito.when(protonSPI.getTransportConnection()).thenReturn(transportConnection);
+      Mockito.when(protonSPI.validateConnection(connection, null)).thenReturn(true);
+
+      ScheduledThreadPoolExecutor scheduledPool = new ScheduledThreadPoolExecutor(
+              ActiveMQDefaultConfiguration.getDefaultScheduledThreadPoolMaxSize());
+
+      AMQPConnectionContext connectionContext = new AMQPConnectionContext(
+              manager,
+              protonSPI,
+              null,
+              (int) ActiveMQClient.DEFAULT_CONNECTION_TTL,
+              manager.getMaxFrameSize(),
+              AMQPConstants.Connection.DEFAULT_CHANNEL_MAX,
+              false,
+              scheduledPool,
+              false,
+              null,
+              null,
+              null,
+              null);
+
+      assertEquals(expectedMaxTransfersPerDelivery, connectionContext.getHandler().getTransport().getMaxTransfersPerDelivery());
+
+      connectionContext.onRemoteOpen(connection);
+      assertEquals(null, connection.getRemoteProperties().get("idk"));
+      connectionContext.close(null);
+   }
 }

docs/user-manual/configuring-transports.adoc

@@ -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.