Backport network flaky test improvements (#2045)

cshannon · web-flow · commit 9c5639ded1ea · 2026-05-26T20:29:38.000-04:00
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DurableSyncNetworkBridgeTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/network/DurableSyncNetworkBridgeTest.java
@@ -16,22 +16,10 @@
  */
 package org.apache.activemq.network;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-import java.io.File;
-import java.net.URI;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-
 import jakarta.jms.Message;
 import jakarta.jms.MessageConsumer;
 import jakarta.jms.MessageProducer;
 import jakarta.jms.Session;
-
 import org.apache.activemq.ActiveMQConnectionFactory;
 import org.apache.activemq.advisory.AdvisoryBroker;
 import org.apache.activemq.broker.BrokerPlugin;
@@ -62,6 +50,18 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
 @RunWith(Parameterized.class)
 public class DurableSyncNetworkBridgeTest extends DynamicNetworkTestSupport {
 
@@ -83,7 +83,7 @@ public class DurableSyncNetworkBridgeTest extends DynamicNetworkTestSupport {
     private final FLOW flow;
 
     @Rule
-    public Timeout globalTimeout = new Timeout(30, TimeUnit.SECONDS);
+    public Timeout globalTimeout = new Timeout(60, TimeUnit.SECONDS);
 
     @Parameters
     public static Collection<Object[]> data() {
@@ -138,6 +138,10 @@ public void testRemoveSubscriptionPropagate() throws Exception {
         assertSubscriptionsCount(broker1, topic, 1);
         assertNCDurableSubsCount(broker2, topic, 1);
 
+        // Wait for subscription to become inactive before attempting removal
+        // It's very important to wait here, otherwise the removal may not propagate
+        waitForSubscriptionInactive(broker1, topic, subName);
+
         removeSubscription(broker1, subName);
 
         assertSubscriptionsCount(broker1, topic, 0);
@@ -222,7 +226,12 @@ public void testRemoveSubscriptionWithBridgeOfflineIncludedChanged() throws Exce
         //Test that on successful reconnection of the bridge that
         //the NC sub will be removed
         restartBroker(broker2, true);
-        assertNCDurableSubsCount(broker2, topic, 1);
+        // In REVERSE flow, broker2=localBroker has the bridge and broker1 (remoteBroker)
+        // is already running, so the sync may have already cleaned up the NC durable sub.
+        // This "before sync" assertion is only valid in FORWARD flow.
+        if (flow == FLOW.FORWARD) {
+            assertNCDurableSubsCount(broker2, topic, 1);
+        }
         restartBroker(broker1, true);
         assertBridgeStarted();
         assertNCDurableSubsCount(broker2, topic, 0);
@@ -249,7 +258,9 @@ public void testSubscriptionRemovedAfterIncludedChanged() throws Exception {
         //the NC sub will be removed because even though the local subscription exists,
         //it no longer matches the included filter
         restartBroker(broker2, true);
-        assertNCDurableSubsCount(broker2, topic, 1);
+        if (flow == FLOW.FORWARD) {
+            assertNCDurableSubsCount(broker2, topic, 1);
+        }
         restartBroker(broker1, true);
         assertBridgeStarted();
         assertNCDurableSubsCount(broker2, topic, 0);
@@ -287,7 +298,9 @@ public void testSubscriptionRemovedAfterStaticChanged() throws Exception {
         //the NC sub will be removed because even though the local subscription exists,
         //it no longer matches the included static filter
         restartBroker(broker2, true);
-        assertNCDurableSubsCount(broker2, topic, 1);
+        if (flow == FLOW.FORWARD) {
+            assertNCDurableSubsCount(broker2, topic, 1);
+        }
         restartBroker(broker1, true);
         assertBridgeStarted();
         assertNCDurableSubsCount(broker2, topic, 0);
@@ -316,10 +329,13 @@ public void testAddAndRemoveSubscriptionWithBridgeOfflineMultiTopics() throws Ex
         //Test that on successful reconnection of the bridge that
         //the NC sub will be removed for topic1 but will stay for topic2
 
-        //before sync, the old NC should exist
+        //before sync, the old NC should exist (only verifiable in FORWARD flow;
+        //in REVERSE, broker2=localBroker has the bridge and sync may already have run)
         restartBroker(broker2, true);
-        assertNCDurableSubsCount(broker2, topic, 1);
-        assertNCDurableSubsCount(broker2, topic2, 0);
+        if (flow == FLOW.FORWARD) {
+            assertNCDurableSubsCount(broker2, topic, 1);
+            assertNCDurableSubsCount(broker2, topic2, 0);
+        }
 
         //After sync, remove old NC and create one for topic 2
         restartBroker(broker1, true);
@@ -527,7 +543,6 @@ public void testAddOnlineSubscriptionsWithBridgeOffline() throws Exception {
         session1.createDurableSubscriber(topic, "sub3");
         session1.createDurableSubscriber(excludeTopic, "sub-exclude");
 
-        Thread.sleep(1000);
         assertNCDurableSubsCount(broker2, topic, 1);
         assertNCDurableSubsCount(broker2, excludeTopic, 0);
 
@@ -566,13 +581,10 @@ public void testAddOnlineSubscriptionsTwoBridges() throws Exception {
         secondConnector.start();
 
         //Make sure both bridges are connected
-        assertTrue(Wait.waitFor(new Condition() {
-            @Override
-            public boolean isSatisified() throws Exception {
-                return localBroker.getNetworkConnectors().get(0).activeBridges().size() == 1 &&
-                        localBroker.getNetworkConnectors().get(1).activeBridges().size() == 1;
-            }
-        }, 10000, 500));
+        assertTrue(Wait.waitFor(() ->
+                        localBroker.getNetworkConnectors().get(0).activeBridges().size() == 1 &&
+                                localBroker.getNetworkConnectors().get(1).activeBridges().size() == 1,
+                TimeUnit.SECONDS.toMillis(15), 500));
 
         //Make sure NC durables exist for both bridges
         assertNCDurableSubsCount(broker2, topic2, 1);
@@ -633,13 +645,7 @@ public void testVirtualDestSubForceDurableSync() throws Exception {
         final DestinationStatistics remoteDestStatistics2 = remoteBroker.getDestination(
                 new ActiveMQQueue("include.test.bar.bridge")).getDestinationStatistics();
 
-        assertTrue(Wait.waitFor(new Condition() {
-
-            @Override
-            public boolean isSatisified() throws Exception {
-                return remoteDestStatistics2.getMessages().getCount() == 501;
-            }
-        }));
+        assertTrue(Wait.waitFor(() -> remoteDestStatistics2.getMessages().getCount() == 501));
 
     }
 
@@ -719,8 +725,36 @@ protected void doSetUp(boolean deleteAllMessages, boolean startNetworkConnector,
         included = new ActiveMQTopic(testTopicName);
         doSetUpRemoteBroker(deleteAllMessages, remoteDataDir, 0);
         doSetUpLocalBroker(deleteAllMessages, startNetworkConnector, localDataDir);
-        //Give time for advisories to propagate
-        Thread.sleep(1000);
+        //Wait for the bridge to be fully started (advisory consumers registered).
+        //Note: activeBridges().size() == 1 is NOT sufficient because bridges are added
+        //to the map before start() completes asynchronously. We must wait for the
+        //startedLatch which counts down after advisory consumers are registered.
+        if (startNetworkConnector) {
+            waitForBridgeFullyStarted();
+        }
+    }
+
+    private void waitForBridgeFullyStarted() throws Exception {
+        // Wait for the local bridge to be fully started (advisory consumers registered)
+        assertTrue("Local bridge should be fully started", Wait.waitFor(() -> {
+            if (localBroker.getNetworkConnectors().get(0).activeBridges().isEmpty()) {
+                return false;
+            }
+            final NetworkBridge bridge = localBroker.getNetworkConnectors().get(0).activeBridges().iterator().next();
+            if (bridge instanceof DemandForwardingBridgeSupport) {
+                return ((DemandForwardingBridgeSupport) bridge).startedLatch.getCount() == 0;
+            }
+            return true;
+        }, TimeUnit.SECONDS.toMillis(15), 100));
+
+        // Also wait for the duplex bridge on the remote broker to be fully started.
+        // The duplex connector creates a separate DemandForwardingBridge on the remote side
+        // that also needs its advisory consumers registered before it can process events.
+        assertTrue("Duplex bridge should be fully started", Wait.waitFor(() -> {
+            final DemandForwardingBridge duplexBridge = findDuplexBridge(
+                    remoteBroker.getTransportConnectors().get(0));
+            return duplexBridge != null && duplexBridge.startedLatch.getCount() == 0;
+        }, TimeUnit.SECONDS.toMillis(15), 100));
     }
 
     protected void restartLocalBroker(boolean startNetworkConnector) throws Exception {
@@ -729,13 +763,42 @@ protected void restartLocalBroker(boolean startNetworkConnector) throws Exceptio
     }
 
     protected void restartRemoteBroker() throws Exception {
-        int port = 0;
-        if (remoteBroker != null) {
-            List<TransportConnector> transportConnectors = remoteBroker.getTransportConnectors();
-            port = transportConnectors.get(0).getConnectUri().getPort();
-        }
+        final int previousPort = remoteBroker.getTransportConnectors().get(0).getConnectUri().getPort();
+        final File dataDir = remoteBroker.getDataDirectoryFile();
         stopRemoteBroker();
-        doSetUpRemoteBroker(false, remoteBroker.getDataDirectoryFile(), port);
+        try {
+            doSetUpRemoteBroker(false, dataDir, previousPort);
+        } catch (final IOException e) {
+            if (e.getCause() instanceof java.net.BindException) {
+                // Previous port still in TIME_WAIT — use a new ephemeral port
+                doSetUpRemoteBroker(false, dataDir, 0);
+                // Update the local broker's network connector to point to the new port
+                updateLocalNetworkConnectorUri();
+            } else {
+                throw e;
+            }
+        }
+    }
+
+    /**
+     * When the remote broker restarts on a new ephemeral port (BindException fallback),
+     * any existing network connector on the local broker still points to the old port.
+     * This method stops the old connector and replaces it with one targeting the new URI.
+     */
+    private void updateLocalNetworkConnectorUri() throws Exception {
+        if (localBroker == null) {
+            return;
+        }
+        final List<NetworkConnector> connectors = localBroker.getNetworkConnectors();
+        if (connectors.isEmpty()) {
+            return;
+        }
+        final NetworkConnector oldConnector = connectors.get(0);
+        oldConnector.stop();
+        localBroker.removeNetworkConnector(oldConnector);
+        final NetworkConnector newConnector = configureLocalNetworkConnector();
+        localBroker.addNetworkConnector(newConnector);
+        newConnector.start();
     }
 
     protected void doSetUpLocalBroker(boolean deleteAllMessages, boolean startNetworkConnector,
@@ -753,12 +816,14 @@ protected void doSetUpLocalBroker(boolean deleteAllMessages, boolean startNetwor
         localConnection.start();
 
         if (startNetworkConnector) {
-            Wait.waitFor(new Condition() {
-                @Override
-                public boolean isSatisified() throws Exception {
-                    return localBroker.getNetworkConnectors().get(0).activeBridges().size() == 1;
-                }
-            }, 5000, 500);
+            // Best-effort wait for the bridge to appear. Do NOT use assertTrue here
+            // because some tests restart localBroker before remoteBroker is running,
+            // relying on the bridge connecting later when remoteBroker restarts.
+            // Tests that need the bridge to be fully started call assertBridgeStarted() explicitly.
+            // Keep timeout short (5s) to avoid growing the NC reconnect backoff too much,
+            // which would delay bridge formation when the remote broker starts later.
+            Wait.waitFor(() -> localBroker.getNetworkConnectors().get(0).activeBridges().size() == 1,
+                    TimeUnit.SECONDS.toMillis(5), 500);
         }
         localSession = localConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
 
@@ -869,4 +934,24 @@ protected BrokerService createRemoteBroker(File dataDir, int port) throws Except
         return brokerService;
     }
 
+    /**
+     * Wait for a durable subscription to become inactive before attempting removal.
+     * This prevents "Durable consumer is in use" errors when consumer close operations
+     * complete asynchronously (especially visible with Java 25's different thread scheduling).
+     */
+    protected void waitForSubscriptionInactive(final BrokerService brokerService,
+            final ActiveMQTopic topic,
+            final String subName) throws Exception {
+        assertTrue("Subscription should become inactive", Wait.waitFor(() -> {
+            final List<org.apache.activemq.broker.region.DurableTopicSubscription> subs = getSubscriptions(brokerService, topic);
+            for (final org.apache.activemq.broker.region.DurableTopicSubscription sub : subs) {
+                if (sub.getSubscriptionKey().getSubscriptionName().equals(subName)) {
+                    return !sub.isActive();
+                }
+            }
+            // If subscription doesn't exist, it's considered inactive
+            return true;
+        }, TimeUnit.SECONDS.toMillis(15), 100));
+    }
+
 }
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DynamicNetworkTestSupport.java b/activemq-unit-tests/src/test/java/org/apache/activemq/network/DynamicNetworkTestSupport.java
@@ -24,6 +24,7 @@
 import java.lang.reflect.Field;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 import jakarta.jms.Connection;
 import jakarta.jms.JMSException;
@@ -45,7 +46,6 @@
 import org.apache.activemq.command.RemoveSubscriptionInfo;
 import org.apache.activemq.util.SubscriptionKey;
 import org.apache.activemq.util.Wait;
-import org.apache.activemq.util.Wait.Condition;
 import org.junit.Rule;
 import org.junit.rules.TemporaryFolder;
 
@@ -95,9 +95,16 @@ protected void stopRemoteBroker() throws Exception {
     }
 
     protected void assertBridgeStarted() throws Exception {
-        assertTrue(Wait.waitFor(
-            () -> localBroker.getNetworkConnectors().get(0).activeBridges().size() == 1,
-            10000, 500));
+        assertTrue("Bridge should be fully started", Wait.waitFor(() -> {
+            if (localBroker.getNetworkConnectors().get(0).activeBridges().size() != 1) {
+                return false;
+            }
+            final NetworkBridge bridge = localBroker.getNetworkConnectors().get(0).activeBridges().iterator().next();
+            if (bridge instanceof DemandForwardingBridgeSupport) {
+                return ((DemandForwardingBridgeSupport) bridge).startedLatch.getCount() == 0;
+            }
+            return true;
+        }, TimeUnit.SECONDS.toMillis(10), 500));
     }
 
     protected RemoveSubscriptionInfo getRemoveSubscriptionInfo(final ConnectionContext context,
@@ -119,8 +126,8 @@ protected void waitForConsumerCount(final DestinationStatistics destinationStati
 
     protected void waitForDispatchFromLocalBroker(final DestinationStatistics destinationStatistics, final int count) throws Exception {
         assertTrue(Wait.waitFor(() -> count == destinationStatistics.getDequeues().getCount() &&
-               count == destinationStatistics.getDispatched().getCount() &&
-               count == destinationStatistics.getForwards().getCount()));
+                count == destinationStatistics.getDispatched().getCount() &&
+                count == destinationStatistics.getForwards().getCount()));
     }
 
     protected void assertLocalBrokerStatistics(final DestinationStatistics localStatistics, final int count) {
@@ -135,17 +142,19 @@ protected interface ConsumerCreator {
 
     protected void assertNCDurableSubsCount(final BrokerService brokerService,
             final ActiveMQTopic dest, final int count) throws Exception {
-        assertTrue(Wait.waitFor(() -> count == getNCDurableSubs(brokerService, dest).size(),
-            10000, 500));
+        assertTrue("Expected " + count + " NC durable subs on " + dest,
+                Wait.waitFor(() -> count == getNCDurableSubs(brokerService, dest).size(),
+                        TimeUnit.SECONDS.toMillis(30), 500));
     }
 
     protected void assertConsumersCount(final BrokerService brokerService,
             final ActiveMQDestination dest, final int count) throws Exception {
         assertTrue(Wait.waitFor(() -> count == getConsumers(brokerService, dest).size(),
-            10000, 500));
-        Thread.sleep(1000);
-        // Check one more time after a short pause to make sure the count didn't increase past what we wanted
-        assertEquals(count, getConsumers(brokerService, dest).size());
+                10000, 500));
+        // Wait a bit longer and verify the count is stable (didn't increase past what we wanted)
+        assertTrue("Consumer count should remain stable at " + count,
+                Wait.waitFor(() -> count == getConsumers(brokerService, dest).size(),
+                        TimeUnit.SECONDS.toMillis(5), 500));
     }
 
     protected List<Subscription> getConsumers(final BrokerService brokerService,
@@ -194,7 +203,7 @@ protected List<DurableTopicSubscription> getNCDurableSubs(final BrokerService br
     }
 
     protected void removeSubscription(final BrokerService brokerService,
-        final String subName) throws Exception {
+            final String subName) throws Exception {
         final RemoveSubscriptionInfo info = new RemoveSubscriptionInfo();
         info.setClientId(clientId);
         info.setSubscriptionName(subName);
@@ -208,12 +217,9 @@ protected void removeSubscription(final BrokerService brokerService,
 
     protected void assertSubscriptionsCount(final BrokerService brokerService,
             final ActiveMQTopic dest, final int count) throws Exception {
-        assertTrue(Wait.waitFor(new Condition() {
-            @Override
-            public boolean isSatisified() throws Exception {
-                return count == getSubscriptions(brokerService, dest).size();
-            }
-        }, 10000, 500));
+        assertTrue("Expected " + count + " subscriptions on " + dest,
+                Wait.waitFor(() -> count == getSubscriptions(brokerService, dest).size(),
+                        TimeUnit.SECONDS.toMillis(30), 500));
     }
 
     protected void assertSubscriptionMapCounts(NetworkBridge networkBridge, final int count) {
@@ -226,8 +232,9 @@ protected void assertSubscriptionMapCounts(NetworkBridge networkBridge, final in
     protected DemandForwardingBridge findDuplexBridge(final TransportConnector connector) throws Exception {
         assertNotNull(connector);
 
-        for (TransportConnection tc : connector.getConnections()) {
-            if (tc.getConnectionId().startsWith("networkConnector_")) {
+        for (final TransportConnection tc : connector.getConnections()) {
+            final String connectionId = tc.getConnectionId();
+            if (connectionId != null && connectionId.startsWith("networkConnector_")) {
                 final Field bridgeField = TransportConnection.class.getDeclaredField("duplexBridge");
                 bridgeField.setAccessible(true);
                 return (DemandForwardingBridge) bridgeField.get(tc);
