AMQ-9854 Add wait for broker to process disconnect before publishing … (#1646)

* AMQ-9854 Add wait for broker to process disconnect before publishing messages in MQTTTest

* AMQ-9854 Fix MQTTTest to check for no active durable subscribers before publishing messages after disconnect

Author: jeanouii

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java

@@ -1671,6 +1671,12 @@ public void testReceiveMessageSentWhileOffline() throws Exception {
         }
         connectionSub.disconnect();
 
+        // Wait for broker to process disconnect before publishing messages for offline delivery.
+        // Check for no active durable subscribers (works for both regular and virtual topic strategies)
+        assertTrue("Subscription should become inactive",
+                Wait.waitFor(() -> brokerService.getAdminView().getDurableTopicSubscribers().length == 0,
+                        5000, 100));
+
         try {
             for (int j = 0; j < numberOfRuns; j++) {
 
@@ -1690,6 +1696,11 @@ public void testReceiveMessageSentWhileOffline() throws Exception {
                     message.ack();
                 }
                 connectionSub.disconnect();
+
+                // Wait for broker to process disconnect before next iteration publishes
+                assertTrue("Subscription should become inactive",
+                        Wait.waitFor(() -> brokerService.getAdminView().getDurableTopicSubscribers().length == 0,
+                                5000, 100));
             }
         } catch (Exception exception) {
             LOG.error("unexpected exception", exception);
@@ -1722,6 +1733,12 @@ public void testReceiveMessageSentWhileOfflineAndBrokerRestart() throws Exceptio
             connectionSub.disconnect();
         }
 
+        // Wait for broker to process disconnect before publishing messages for offline delivery.
+        // Check for no active durable subscribers (works for both regular and virtual topic strategies)
+        assertTrue("Subscription should become inactive",
+                Wait.waitFor(() -> brokerService.getAdminView().getDurableTopicSubscribers().length == 0,
+                        5000, 100));
+
         MQTT mqttPubLoop = createMQTTConnection("MQTT-Pub-Client", true);
         BlockingConnection connectionPub = mqttPubLoop.blockingConnection();
         connectionPub.connect();