Multi broker setup hardening (#1627)

* Use ephemeral ports for broker creation in DurableFiveBrokerNetworkBridgeTest to prevent port conflicts during parallel test execution

Enhance broker startup reliability by replacing Thread.sleep with Wait.waitFor for transport connector readiness

* Improve broker readiness check in JmsMultipleBrokersTestSupport to skip VM-only brokers and enhance reliability

* Fix startAllBrokers() transport connector check for VM-only and secured brokers
- Skip readiness check for VM transport connectors (no network to verify)
- Treat JMSSecurityException as ready (broker accepts connections, just enforces auth)
- Fix AdvisoryViaNetworkTest race conditions with waitForConsumerOnBroker()

Author: jeanouii

activemq-unit-tests/src/test/java/org/apache/activemq/JmsMultipleBrokersTestSupport.java

@@ -271,14 +271,46 @@ protected void waitForBridgeFormation() throws Exception {
     }
 
     protected void startAllBrokers() throws Exception {
-        Collection<BrokerItem> brokerList = brokers.values();
+        final Collection<BrokerItem> brokerList = brokers.values();
         for (Iterator<BrokerItem> i = brokerList.iterator(); i.hasNext();) {
-            BrokerService broker = i.next().broker;
+            final BrokerService broker = i.next().broker;
             broker.start();
             broker.waitUntilStarted();
         }
 
-        Thread.sleep(maxSetupTime);
+        // Wait for all brokers with TCP transport connectors to be ready to accept connections
+        // instead of using Thread.sleep which is unreliable across different machines.
+        // Skip this check for VM-only brokers since VM transport doesn't require network readiness
+        // and some tests intentionally configure brokers to reject connections.
+        for (final BrokerItem brokerItem : brokerList) {
+            final BrokerService broker = brokerItem.broker;
+            // Only check transport connector readiness for TCP/network connectors
+            // (skip VM-only brokers which don't require network readiness checks)
+            final boolean hasTcpConnector = broker.getTransportConnectors().stream()
+                .anyMatch(tc -> {
+                    final String scheme = tc.getUri().getScheme();
+                    return scheme != null && !scheme.equals("vm");
+                });
+            if (hasTcpConnector) {
+                assertTrue("Broker " + broker.getBrokerName() + " transport connectors ready",
+                    Wait.waitFor(() -> {
+                        if (!broker.isStarted()) {
+                            return false;
+                        }
+                        // Try to create a test connection to verify transport is accepting connections
+                        try (final Connection testConn = brokerItem.createConnection()) {
+                            return true;
+                        } catch (final jakarta.jms.JMSSecurityException e) {
+                            // Security exception means the broker IS accepting connections,
+                            // it's just enforcing authentication - consider it ready
+                            return true;
+                        } catch (final Exception e) {
+                            LOG.debug("Broker {} not ready yet: {}", broker.getBrokerName(), e.getMessage());
+                            return false;
+                        }
+                    }, TimeUnit.SECONDS.toMillis(30), TimeUnit.SECONDS.toMillis(1)));
+            }
+        }
     }
 
     protected BrokerService createBroker(String brokerName) throws Exception {

activemq-unit-tests/src/test/java/org/apache/activemq/network/DurableFiveBrokerNetworkBridgeTest.java

@@ -313,9 +313,8 @@ protected void testDurablePropagation5Broker() throws Exception {
         // Setup consumers
         Session ses = createSession("Broker_A_A");
         MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA");
-        Thread.sleep(1000);
 
-        // let consumers propagate around the network
+        // let consumers propagate around the network (assertNCDurableSubsCount waits internally)
         assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 1);
         assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
         assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 1);
@@ -328,9 +327,8 @@ protected void testDurablePropagation5Broker() throws Exception {
         //bring online a consumer on the other side
         Session ses2 = createSession("Broker_E_E");
         MessageConsumer clientE = ses2.createDurableSubscriber(dest, "subE");
-        Thread.sleep(1000);
 
-        //there will be 2 network durables, 1 for each direction of the bridge
+        //there will be 2 network durables, 1 for each direction of the bridge (assertNCDurableSubsCount waits internally)
         assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2);
         assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 2);
         assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 2);
@@ -384,16 +382,14 @@ protected void testDurablePropagationSpoke() throws Exception {
 
         MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA");
         MessageConsumer clientAB = ses.createDurableSubscriber(dest, "subAB");
-        Thread.sleep(1000);
 
-        // let consumers propagate around the network
+        // let consumers propagate around the network (assertNCDurableSubsCount waits internally)
         assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 1);
         assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
         assertNCDurableSubsCount(brokers.get("Broker_D_D").broker, dest, 1);
         assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);
 
         MessageConsumer clientD = ses4.createDurableSubscriber(dest, "subD");
-        Thread.sleep(1000);
 
         assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 2);
         assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, 1);
@@ -759,7 +755,6 @@ private void testDurablePropagation(int ttl, boolean dynamicOnly, boolean restar
         Session ses2 = createSession("Broker_E_E");
         MessageConsumer clientA = ses.createDurableSubscriber(dest, "subA");
         MessageConsumer clientE = ses2.createDurableSubscriber(dest, "subE");
-        Thread.sleep(1000);
 
         assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, 0);
         assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, 0);
@@ -768,9 +763,8 @@ private void testDurablePropagation(int ttl, boolean dynamicOnly, boolean restar
         assertNCDurableSubsCount(brokers.get("Broker_E_E").broker, dest, 0);
 
         startNetworkConnectors(nc1, nc2, nc3, nc4);
-        Thread.sleep(1000);
 
-        // Check that the correct network durables exist
+        // Check that the correct network durables exist (assertNCDurableSubsCount waits internally)
         assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, expected.get(0));
         assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, expected.get(1));
         assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, expected.get(2));
@@ -795,12 +789,10 @@ private void testDurablePropagation(int ttl, boolean dynamicOnly, boolean restar
             // to test sync works ok. Things should work for all cases both dynamicOnly
             // false and true because TTL info still exits and consumers are online
             stopNetworkConnectors(nc1, nc2, nc3, nc4);
-            Thread.sleep(1000);
             startNetworkConnectors(nc1, nc2, nc3, nc4);
-            Thread.sleep(1000);
         }
 
-        // after restarting the bridges, check sync/demand are correct
+        // after restarting the bridges, check sync/demand are correct (assertNCDurableSubsCount waits internally)
         assertNCDurableSubsCount(brokers.get("Broker_A_A").broker, dest, expected.get(0));
         assertNCDurableSubsCount(brokers.get("Broker_B_B").broker, dest, expected.get(1));
         assertNCDurableSubsCount(brokers.get("Broker_C_C").broker, dest, expected.get(2));
@@ -846,12 +838,13 @@ public void setUp() throws Exception {
         super.setAutoFail(true);
         super.setUp();
         deletePersistentMessagesOnStartup = true;
-        String options = new String("?persistent=true&useJmx=false");
-        createBroker(new URI("broker:(tcp://localhost:61616)/Broker_A_A" + options));
-        createBroker(new URI("broker:(tcp://localhost:61617)/Broker_B_B" + options));
-        createBroker(new URI("broker:(tcp://localhost:61618)/Broker_C_C" + options));
-        createBroker(new URI("broker:(tcp://localhost:61619)/Broker_D_D" + options));
-        createBroker(new URI("broker:(tcp://localhost:61620)/Broker_E_E" + options));
+        final String options = "?persistent=true&useJmx=false";
+        // Use ephemeral ports (0) to avoid port conflicts when tests run in parallel
+        createBroker(new URI("broker:(tcp://localhost:0)/Broker_A_A" + options));
+        createBroker(new URI("broker:(tcp://localhost:0)/Broker_B_B" + options));
+        createBroker(new URI("broker:(tcp://localhost:0)/Broker_C_C" + options));
+        createBroker(new URI("broker:(tcp://localhost:0)/Broker_D_D" + options));
+        createBroker(new URI("broker:(tcp://localhost:0)/Broker_E_E" + options));
     }
 
     @Override

activemq-unit-tests/src/test/java/org/apache/activemq/usecases/AdvisoryViaNetworkTest.java

@@ -104,6 +104,10 @@ public void testAdvisoryPrefetchSize() throws Exception {
         createConsumer("A", topic1);
         createConsumer("A", new ActiveMQTopic("A.FOO2"));
 
+        // Wait for network bridge to propagate consumers to broker B
+        waitForConsumerOnBroker(brokerB, advisoryTopic);
+        waitForConsumerOnBroker(brokerB, topic1);
+
         //verify that brokerB's advisory prefetch is 10 but normal topic prefetch is 1
         assertEquals(10, brokerB.getDestination(advisoryTopic).getConsumers().get(0).getPrefetchSize());
         assertEquals(1, brokerB.getDestination(topic1).getConsumers().get(0).getPrefetchSize());
@@ -137,6 +141,10 @@ public void testAdvisoryPrefetchSize1() throws Exception {
         createConsumer("A", topic1);
         createConsumer("A", new ActiveMQTopic("A.FOO2"));
 
+        // Wait for network bridge to propagate consumers to broker B
+        waitForConsumerOnBroker(brokerB, advisoryTopic);
+        waitForConsumerOnBroker(brokerB, topic1);
+
         //verify that brokerB's advisory prefetch is 1 but normal topic prefetch is 10
         assertEquals(1, brokerB.getDestination(advisoryTopic).getConsumers().get(0).getPrefetchSize());
         assertEquals(10, brokerB.getDestination(topic1).getConsumers().get(0).getPrefetchSize());
@@ -168,6 +176,10 @@ public void testAdvisoryPrefetchSizeNotSet() throws Exception {
         createConsumer("A", topic1);
         createConsumer("A", new ActiveMQTopic("A.FOO2"));
 
+        // Wait for network bridge to propagate consumers to broker B
+        waitForConsumerOnBroker(brokerB, advisoryTopic);
+        waitForConsumerOnBroker(brokerB, topic1);
+
         //verify that both consumers have a prefetch of 10
         assertEquals(10, brokerB.getDestination(advisoryTopic).getConsumers().get(0).getPrefetchSize());
         assertEquals(10, brokerB.getDestination(topic1).getConsumers().get(0).getPrefetchSize());
@@ -198,6 +210,10 @@ public void testPrefetchSize1() throws Exception {
         createConsumer("A", topic1);
         createConsumer("A", new ActiveMQTopic("A.FOO2"));
 
+        // Wait for network bridge to propagate consumers to broker B
+        waitForConsumerOnBroker(brokerB, advisoryTopic);
+        waitForConsumerOnBroker(brokerB, topic1);
+
         //verify that both consumers have a prefetch of 1
         assertEquals(1, brokerB.getDestination(advisoryTopic).getConsumers().get(0).getPrefetchSize());
         assertEquals(1, brokerB.getDestination(topic1).getConsumers().get(0).getPrefetchSize());
@@ -392,4 +408,15 @@ public void setUp() throws Exception {
         super.setUp();
     }
 
+    private void waitForConsumerOnBroker(final BrokerService broker, final ActiveMQTopic topic) throws Exception {
+        assertTrue("Consumer on " + broker.getBrokerName() + " for " + topic,
+            Wait.waitFor(() -> {
+                try {
+                    return !broker.getDestination(topic).getConsumers().isEmpty();
+                } catch (final Exception e) {
+                    return false;
+                }
+            }, 30000, 100));
+    }
+
 }