You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by jl...@apache.org on 2026/03/10 21:49:17 UTC
(activemq) branch main updated: fix(test): fix MQTTTest#testReceiveMessageSentWhileOffline flakiness (#1751)
This is an automated email from the ASF dual-hosted git repository.
jlmonteiro pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/main by this push:
new 1f3b4cc017 fix(test): fix MQTTTest#testReceiveMessageSentWhileOffline flakiness (#1751)
1f3b4cc017 is described below
commit 1f3b4cc017c8c1ded35301d1ae18c6ea837c2181
Author: JB Onofré <jb...@apache.org>
AuthorDate: Tue Mar 10 22:49:11 2026 +0100
fix(test): fix MQTTTest#testReceiveMessageSentWhileOffline flakiness (#1751)
* fix(test): fix MQTTTest#testReceiveMessageSentWhileOffline flakiness
subscribe() returns on SUBACK but the broker processes ConsumerInfo
asynchronously, so the durable subscription may not be fully reactivated
when receive() is called. Add Wait.waitFor on isSubscriptionActive()
after subscribe to ensure the broker has activated the subscription
before attempting to receive queued messages.
Also remove the try/catch that silently swallowed assertion failures
mid-loop, which masked the real failure with a misleading count mismatch.
* fix(test): increase subscription activation timeout to 30s in MQTTTest
The NIO+SSL transport variant (MQTTNIOSSLTest) can be slower to
reactivate subscriptions under CI load, causing the 15s timeout to
be insufficient.
---
.../apache/activemq/transport/mqtt/MQTTTest.java | 68 +++++++++++++---------
1 file changed, 42 insertions(+), 26 deletions(-)
diff --git a/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java b/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java
index 1f95b03e8d..6729a968d7 100644
--- a/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java
+++ b/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java
@@ -1640,7 +1640,7 @@ public class MQTTTest extends MQTTTestSupport {
assertEquals("Should have received " + topics.length + " messages", topics.length, received);
}
- @Test(timeout = 60 * 1000)
+ @Test(timeout = 2 * 60 * 1000)
public void testReceiveMessageSentWhileOffline() throws Exception {
final byte[] payload = new byte[1024 * 32];
for (int i = 0; i < payload.length; i++) {
@@ -1679,36 +1679,38 @@ public class MQTTTest extends MQTTTestSupport {
// Wait for broker to process disconnect before publishing messages for offline delivery.
assertTrue("Subscription should become inactive",
Wait.waitFor(() -> isSubscriptionInactive(topics[0], mqttSub.getClientId().toString()),
- TimeUnit.SECONDS.toMillis(5), 100));
-
- try {
- for (int j = 0; j < numberOfRuns; j++) {
+ TimeUnit.SECONDS.toMillis(10), 100));
- for (int i = 0; i < messagesPerRun; ++i) {
- connectionPub.publish(topics[0].name().toString(), payload, QoS.AT_LEAST_ONCE, false);
- }
+ for (int j = 0; j < numberOfRuns; j++) {
- connectionSub = mqttSub.blockingConnection();
- connectionSub.connect();
- connectionSub.subscribe(topics);
+ for (int i = 0; i < messagesPerRun; ++i) {
+ connectionPub.publish(topics[0].name().toString(), payload, QoS.AT_LEAST_ONCE, false);
+ }
- for (int i = 0; i < messagesPerRun; ++i) {
- Message message = connectionSub.receive(5, TimeUnit.SECONDS);
- assertNotNull(message);
- received++;
- assertTrue(Arrays.equals(payload, message.getPayload()));
- message.ack();
- }
- connectionSub.disconnect();
+ connectionSub = mqttSub.blockingConnection();
+ connectionSub.connect();
+ connectionSub.subscribe(topics);
- // Wait for broker to process disconnect before next iteration publishes
- assertTrue("Subscription should become inactive",
- Wait.waitFor(() -> isSubscriptionInactive(topics[0], mqttSub.getClientId().toString()),
- TimeUnit.SECONDS.toMillis(5), 100));
+ // Wait for broker to fully activate the subscription and start dispatching
+ // queued messages. subscribe() returns on SUBACK but broker processes the
+ // ConsumerInfo asynchronously, so messages may not be ready for dispatch yet.
+ assertTrue("Subscription should become active in run " + (j + 1),
+ Wait.waitFor(() -> isSubscriptionActive(topics[0], mqttSub.getClientId().toString()),
+ TimeUnit.SECONDS.toMillis(30), 100));
+
+ for (int i = 0; i < messagesPerRun; ++i) {
+ Message message = connectionSub.receive(5, TimeUnit.SECONDS);
+ assertNotNull("Should have received message " + (i + 1) + " of " + messagesPerRun + " in run " + (j + 1), message);
+ received++;
+ assertTrue(Arrays.equals(payload, message.getPayload()));
+ message.ack();
}
- } catch (Exception exception) {
- LOG.error("unexpected exception", exception);
- exception.printStackTrace();
+ connectionSub.disconnect();
+
+ // Wait for broker to process disconnect before next iteration publishes
+ assertTrue("Subscription should become inactive",
+ Wait.waitFor(() -> isSubscriptionInactive(topics[0], mqttSub.getClientId().toString()),
+ TimeUnit.SECONDS.toMillis(10), 100));
}
assertEquals("Should have received " + (messagesPerRun * (numberOfRuns + 1)) + " messages", (messagesPerRun * (numberOfRuns + 1)), received);
}
@@ -1727,6 +1729,20 @@ public class MQTTTest extends MQTTTestSupport {
}
}
+ private boolean isSubscriptionActive(Topic topic, String clientId) throws Exception {
+ if (isVirtualTopicSubscriptionStrategy()) {
+ String queueName = buildVirtualTopicQueueName(topic, clientId);
+ try {
+ return getProxyToQueue(queueName).getConsumerCount() > 0;
+ } catch (Exception ignore) {
+ return false;
+ }
+ } else {
+ return brokerService.getAdminView().getDurableTopicSubscribers().length == 1 &&
+ brokerService.getAdminView().getInactiveDurableTopicSubscribers().length == 0;
+ }
+ }
+
private boolean isVirtualTopicSubscriptionStrategy() {
String config = getProtocolConfig();
return config != null && config.contains("mqtt-virtual-topic-subscriptions");
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@activemq.apache.org
For additional commands, e-mail: commits-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact