fix(test): resolve flaky tests (#1774)

* fix(test): resolve flaky tests

* fix(test): try to make it more stable

Author: jeanouii

activemq-unit-tests/src/test/java/org/apache/activemq/broker/advisory/AdvisoryBrokerTest.java

@@ -285,34 +285,41 @@ public void testProducerAdvisoriesReplayedOnlyTargetNewConsumer() throws Excepti
 
         ActiveMQDestination queue = new ActiveMQQueue("test");
         ActiveMQDestination destination = AdvisorySupport.getProducerAdvisoryTopic(queue);
-        
+
         // Setup a first connection
         StubConnection connection1 = createConnection();
         ConnectionInfo connectionInfo1 = createConnectionInfo();
         SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
         connection1.send(connectionInfo1);
         connection1.send(sessionInfo1);
-        // Create the first consumer..         
+        // Create the first consumer..
+        // Use request() to ensure the advisory consumer is fully registered before
+        // the producer is created. This prevents a race where addConsumer and addProducer
+        // execute concurrently on different transport threads, which could cause the
+        // advisory consumer to receive both a broadcast AND a replay of the same producer.
         ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
         consumerInfo1.setPrefetchSize(100);
-        connection1.send(consumerInfo1);
+        connection1.request(consumerInfo1);
 
         // Setup a producer.
         StubConnection connection2 = createConnection();
         ConnectionInfo connectionInfo2 = createConnectionInfo();
         SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2);
         ProducerInfo producerInfo2 = createProducerInfo(sessionInfo2);
-        producerInfo2.setDestination(queue);        
+        producerInfo2.setDestination(queue);
         connection2.send(connectionInfo2);
         connection2.send(sessionInfo2);
-        connection2.send(producerInfo2);
-        
+        // Use request() to ensure producer is fully registered and its "new producer"
+        // advisory is fired before we create the advisory consumer. This prevents a race
+        // where the advisory consumer could receive both a replay AND the broadcast advisory.
+        connection2.request(producerInfo2);
+
         Message m1 = receiveMessage(connection1);
         assertNotNull(m1);
         assertNotNull(m1.getDataStructure());
         assertEquals(((ProducerInfo)m1.getDataStructure()).getProducerId(), producerInfo2.getProducerId());
-        
-        // Create the 2nd consumer..         
+
+        // Create the 2nd consumer..
         ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination);
         consumerInfo2.setPrefetchSize(100);
         connection2.send(consumerInfo2);

activemq-unit-tests/src/test/java/org/apache/activemq/broker/scheduler/JmsCronSchedulerTest.java

@@ -30,9 +30,7 @@
 
 import jakarta.jms.Connection;
 import jakarta.jms.JMSException;
-import jakarta.jms.Message;
 import jakarta.jms.MessageConsumer;
-import jakarta.jms.MessageListener;
 import jakarta.jms.MessageProducer;
 import jakarta.jms.Session;
 import jakarta.jms.TextMessage;
@@ -57,19 +55,16 @@ public void testSimulatenousCron() throws Exception {
         MessageConsumer consumer = session.createConsumer(destination);
 
         final CountDownLatch latch = new CountDownLatch(COUNT);
-        consumer.setMessageListener(new MessageListener() {
-            @Override
-            public void onMessage(Message message) {
-                count.incrementAndGet();
-                latch.countDown();
-                assertTrue(message instanceof TextMessage);
-                TextMessage tm = (TextMessage) message;
-                try {
-                    LOG.info("Received [{}] count: {} ", tm.getText(), count.get());
-                } catch (JMSException e) {
-                    LOG.error("Unexpected exception in onMessage", e);
-                    fail("Unexpected exception in onMessage: " + e.getMessage());
-                }
+        consumer.setMessageListener(message -> {
+            count.incrementAndGet();
+            latch.countDown();
+            assertTrue(message instanceof TextMessage);
+            final TextMessage tm = (TextMessage) message;
+            try {
+                LOG.info("Received [{}] count: {} ", tm.getText(), count.get());
+            } catch (JMSException e) {
+                LOG.error("Unexpected exception in onMessage", e);
+                fail("Unexpected exception in onMessage: " + e.getMessage());
             }
         });
 
@@ -88,9 +83,11 @@ public void onMessage(Message message) {
         JobScheduler js = sb.getJobScheduler();
         List<Job> list = js.getAllJobs();
         assertEquals(COUNT, list.size());
-        latch.await(2, TimeUnit.MINUTES);
-        // All should messages should have been received by now
-        assertEquals(COUNT, count.get());
+        assertTrue("all scheduled messages should fire", latch.await(2, TimeUnit.MINUTES));
+        // Cron "* * * * *" fires every minute, so count may exceed COUNT
+        // if a second minute boundary is crossed during the wait
+        assertTrue("at least " + COUNT + " messages received, got " + count.get(),
+                count.get() >= COUNT);
 
         connection.close();
     }

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

@@ -803,11 +803,11 @@ private void testDurablePropagation(final int ttl, final boolean dynamicOnly, fi
 
     protected void assertNCDurableSubsCount(final BrokerService brokerService, final ActiveMQTopic dest,
             final int count) throws Exception {
+        final boolean result = Wait.waitFor(() -> count == getNCDurableSubs(brokerService, dest).size(),
+                TimeUnit.SECONDS.toMillis(30), 500);
         assertTrue("Expected " + count + " NC durable sub(s) on " + brokerService.getBrokerName()
                 + " for " + dest.getTopicName() + ", but got "
-                + getNCDurableSubs(brokerService, dest).size(),
-            Wait.waitFor(() -> count == getNCDurableSubs(brokerService, dest).size(),
-                TimeUnit.SECONDS.toMillis(30), 500));
+                + getNCDurableSubs(brokerService, dest).size(), result);
     }
 
     protected List<DurableTopicSubscription> getNCDurableSubs(final BrokerService brokerService,
@@ -824,7 +824,7 @@ protected List<DurableTopicSubscription> getNCDurableSubs(final BrokerService br
         for (final SubscriptionKey key : destination.getDurableTopicSubs().keySet()) {
             if (key.getSubscriptionName().startsWith(DemandForwardingBridge.DURABLE_SUB_PREFIX)) {
                 final DurableTopicSubscription sub = destination.getDurableTopicSubs().get(key);
-                if (sub != null && sub.isActive()) {
+                if (sub != null) {
                     subs.add(sub);
                 }
             }

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

@@ -97,6 +97,20 @@ protected void testNonDurableReceiveThrougRestart(final String pubBroker, final
         networkConnector.start();
         waitForBridgeFormation();
 
+        // Wait for the network bridge to re-establish its demand subscription on the
+        // publishing broker. waitForBridgeFormation() only verifies the bridge is connected,
+        // but setupStaticDestinations() (which creates the durable demand subscription)
+        // runs asynchronously after bridge connection. Without this wait, sendMessages()
+        // can fire before the demand subscription is set up, causing the message to be
+        // published with no subscriber to forward it to the consumer broker.
+        // We check for an active durable subscription because the durable sub may already
+        // exist (inactive) from the previous bridge; we need it to be reactivated.
+        final Topic pubBrokerDest = (Topic) brokers.get(pubBroker).broker.getDestination(dest);
+        assertTrue("Network durable subscription should be active on " + pubBroker,
+                Wait.waitFor(() -> pubBrokerDest.getDurableTopicSubs().values().stream()
+                                .anyMatch(DurableTopicSubscription::isActive),
+                        10000, 100));
+
         // Send messages
         sendMessages(pubBroker, dest, 1);
 

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

@@ -97,7 +97,7 @@ public void testJMXCountersWithOfflineSubs() throws Exception {
         MessageConsumer consumer = session.createDurableSubscriber(topic, "SubsId", null, true);
 
         for (int i=0; i<sent/2; i++) {
-            Message m =  consumer.receive(4000);
+            final Message m =  consumer.receive(30000);
             assertNotNull("got message: " + i, m);
             LOG.info("Got :" + i + ", " + m);
         }
@@ -110,12 +110,8 @@ public void testJMXCountersWithOfflineSubs() throws Exception {
 
         assertTrue("is active", durableSubscriptionView.isActive());
         assertEquals("all enqueued", keepDurableSubsActive ? 10 : 0, durableSubscriptionView.getEnqueueCounter());
-        assertTrue("correct waiting acks", Wait.waitFor(new Wait.Condition() {
-            @Override
-            public boolean isSatisified() throws Exception {
-                return 5 == durableSubscriptionView.getMessageCountAwaitingAcknowledge();
-            }
-        }));
+        assertTrue("correct waiting acks", Wait.waitFor(
+                () -> 5 == durableSubscriptionView.getMessageCountAwaitingAcknowledge()));
         assertEquals("correct dequeue", 5, durableSubscriptionView.getDequeueCounter());
 
 
@@ -150,7 +146,7 @@ public boolean isSatisified() throws Exception {
         consumer = session.createDurableSubscriber(topic, "SubsId", null, true);
 
         for (int i=0; i<sent/2;i++) {
-            Message m =  consumer.receive(30000);
+            final Message m =  consumer.receive(30000);
             assertNotNull("got message: " + i, m);
             LOG.info("Got :" + i + ", " + m);
         }
@@ -162,13 +158,10 @@ public boolean isSatisified() throws Exception {
 
         assertTrue("is active", durableSubscriptionView2.isActive());
         assertEquals("all enqueued", keepDurableSubsActive ? 10 : 0, durableSubscriptionView2.getEnqueueCounter());
-        assertTrue("correct dequeue", Wait.waitFor(new Wait.Condition() {
-            @Override
-            public boolean isSatisified() throws Exception {
-                long val = durableSubscriptionView2.getDequeueCounter();
-                LOG.info("dequeue count:" + val);
-                return 10 == val;
-            }
+        assertTrue("correct dequeue", Wait.waitFor(() -> {
+            final long val = durableSubscriptionView2.getDequeueCounter();
+            LOG.info("dequeue count:" + val);
+            return 10 == val;
         }));
     }
 }