ARTEMIS-4091 - Make scaleDown target more deterministic

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

@@ -41,6 +41,7 @@ import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
 import org.apache.activemq.artemis.spi.core.remoting.BaseConnectionLifeCycleListener;
 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.Env;
 import org.apache.activemq.artemis.utils.IPV6Util;
 import org.slf4j.Logger;
@@ -436,7 +437,9 @@ public class NettyConnection implements Connection {
             continue;
          }
          if (NettyConnectorFactory.class.getName().equals(cfg.getFactoryClassName())) {
-            if (configuration.get(TransportConstants.PORT_PROP_NAME).equals(cfg.getParams().get(TransportConstants.PORT_PROP_NAME))) {
+            int port1 = ConfigurationHelper.getIntProperty(TransportConstants.PORT_PROP_NAME, TransportConstants.DEFAULT_PORT, configuration);
+            int port2 = ConfigurationHelper.getIntProperty(TransportConstants.PORT_PROP_NAME, TransportConstants.DEFAULT_PORT, cfg.getParams());
+            if (port1 == port2) {
                //port same, check host
                Object hostParam = configuration.get(TransportConstants.HOST_PROP_NAME);
                if (hostParam != null) {

artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java

@@ -646,7 +646,7 @@ public interface ActiveMQServerLogger {
    @LogMessage(id = 222111, value = "exception while invoking {} on {}", level = LogMessage.Level.TRACE)
    void managementOperationError(String op, String resourceName, Exception e);
 
-   @LogMessage(id = 222112, value = "exception while retrieving attribute {} on {}", level = LogMessage.Level.WARN)
+   @LogMessage(id = 222112, value = "exception while retrieving attribute {} on {}", level = LogMessage.Level.TRACE)
    void managementAttributeError(String att, String resourceName, Exception e);
 
    @LogMessage(id = 222113, value = "On ManagementService stop, there are {} unexpected registered MBeans: {}", level = LogMessage.Level.WARN)

artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionBridge.java

@@ -24,6 +24,7 @@ import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
 
 import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
+import org.apache.activemq.artemis.api.core.ActiveMQAddressDoesNotExistException;
 import org.apache.activemq.artemis.api.core.ActiveMQException;
 import org.apache.activemq.artemis.api.core.ActiveMQExceptionType;
 import org.apache.activemq.artemis.api.core.Message;
@@ -430,7 +431,8 @@ public class ClusterConnectionBridge extends BridgeImpl {
             try {
                queue.deleteQueue(true);
                queue.removeAddress();
-
+            } catch (ActiveMQAddressDoesNotExistException e) {
+               // ignore
             } catch (Exception e) {
                logger.warn(e.getMessage(), e);
             }

artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LiveOnlyActivation.java

@@ -149,7 +149,20 @@ public class LiveOnlyActivation extends Activation {
          // a timeout is necessary here in case we use a NamedLiveNodeLocatorForScaleDown and there's no matching node in the cluster
          // should the timeout be configurable?
          nodeLocator.locateNode(ActiveMQClient.DEFAULT_DISCOVERY_INITIAL_WAIT_TIMEOUT);
+
          ClientSessionFactoryInternal clientSessionFactory = null;
+         if (nodeLocator instanceof AnyLiveNodeLocatorForScaleDown && scaleDownPolicy.getConnectors() != null) {
+            try {
+               clientSessionFactory = scaleDownServerLocator.connect();
+            } catch (Exception e) {
+               logger.trace("Failed to connect to {}", scaleDownPolicy.getConnectors().get(0));
+               if (clientSessionFactory != null) {
+                  clientSessionFactory.close();
+               }
+               clientSessionFactory = null;
+            }
+         }
+
          while (clientSessionFactory == null) {
             Pair<TransportConfiguration, TransportConfiguration> possibleLive = null;
             possibleLive = nodeLocator.getLiveConfiguration();

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/ScaleDownDeterminism.java

@@ -0,0 +1,108 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.activemq.artemis.tests.integration.server;
+
+import java.lang.invoke.MethodHandles;
+
+import org.apache.activemq.artemis.api.core.client.ClientProducer;
+import org.apache.activemq.artemis.api.core.client.ClientSession;
+import org.apache.activemq.artemis.core.server.cluster.impl.MessageLoadBalancingType;
+import org.apache.activemq.artemis.tests.integration.cluster.distribution.ClusterTestBase;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ScaleDownDeterminism extends ClusterTestBase {
+
+   private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+   @Override
+   @Before
+   public void setUp() throws Exception {
+      super.setUp();
+      setupLiveServer(0, isFileStorage(), HAType.SharedNothingReplication, isNetty(), true);
+      servers[0].getConfiguration().setSecurityEnabled(true);
+      setupLiveServer(1, isFileStorage(), HAType.SharedNothingReplication, isNetty(), true);
+      servers[1].getConfiguration().setSecurityEnabled(true);
+      setupLiveServer(2, isFileStorage(), HAType.SharedNothingReplication, isNetty(), true);
+      servers[2].getConfiguration().setSecurityEnabled(true);
+
+      setupClusterConnection("cluster0", "testAddress", MessageLoadBalancingType.ON_DEMAND, 1, isNetty(), 0, 1, 2);
+      setupClusterConnection("cluster0", "testAddress", MessageLoadBalancingType.ON_DEMAND, 1, isNetty(), 1, 0, 2);
+      setupClusterConnection("cluster0", "testAddress", MessageLoadBalancingType.ON_DEMAND, 1, isNetty(), 2, 0, 1);
+
+      startServers(0, 1, 2);
+      setupSessionFactory(0, isNetty(), false, servers[0].getConfiguration().getClusterUser(), servers[0].getConfiguration().getClusterPassword());
+      setupSessionFactory(1, isNetty(), false, servers[1].getConfiguration().getClusterUser(), servers[1].getConfiguration().getClusterPassword());
+      setupSessionFactory(2, isNetty(), false, servers[2].getConfiguration().getClusterUser(), servers[2].getConfiguration().getClusterPassword());
+      logger.debug("===============================");
+      logger.debug("Node 0: {}", servers[0].getClusterManager().getNodeId());
+      logger.debug("Node 1: {}", servers[1].getClusterManager().getNodeId());
+      logger.debug("Node 2: {}", servers[2].getClusterManager().getNodeId());
+      logger.debug("===============================");
+
+      servers[0].setIdentity("Node0");
+      servers[1].setIdentity("Node1");
+      servers[2].setIdentity("Node2");
+   }
+
+   protected boolean isNetty() {
+      return true;
+   }
+
+   @Test
+   public void testScaleDownDeterministically() throws Exception {
+      final String queueName = "testQueue";
+      final int messageCount = 10;
+
+      ClientSession session = sfs[0].createSession(servers[0].getConfiguration().getClusterUser(), servers[0].getConfiguration().getClusterPassword(), false, true, false, false, 0);
+      createQueue(0, queueName, queueName, null, false, servers[0].getConfiguration().getClusterUser(), servers[0].getConfiguration().getClusterPassword());
+      ClientProducer producer = session.createProducer(queueName);
+
+      for (int i = 0; i < messageCount; i++) {
+         producer.send(session.createMessage(false));
+      }
+      session.close();
+      sfs[0].close();
+
+      servers[0].getActiveMQServerControl().addConnector("scaleDown", "tcp://localhost:61617");
+      //Connectors set up in test do ot use the host param whicvh is needed for above command
+      //Removing host param so that cluster connector matches new scaleDown connector
+      servers[0].getConfiguration().getConnectorConfigurations().get("scaleDown").getParams().remove("host");
+
+      String server0connector2 = servers[0].getConfiguration().getClusterConfigurations().iterator().next().getStaticConnectors().get(1);
+      String server1connector1 = servers[1].getConfiguration().getClusterConfigurations().iterator().next().getStaticConnectors().get(0);
+
+      servers[0].getActiveMQServerControl().scaleDown("scaleDown");
+      Assert.assertEquals(messageCount, servers[1].getTotalMessageCount());
+      servers[0].start();
+
+      waitForServerToStart(servers[0]);
+      Assert.assertEquals(0, servers[0].getTotalMessageCount());
+
+      servers[1].getActiveMQServerControl().scaleDown(server1connector1);
+      Assert.assertEquals(messageCount, servers[0].getTotalMessageCount());
+      servers[1].start();
+
+      waitForServerToStart(servers[1]);
+      servers[0].getActiveMQServerControl().scaleDown(server0connector2);
+      Assert.assertEquals(messageCount, servers[2].getTotalMessageCount());
+   }
+}