You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@artemis.apache.org by cl...@apache.org on 2026/09/09 16:28:57 UTC
(artemis) 02/15: ARTEMIS-6203 make MQTT resiliency soak tests more robust
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch 2.57.x
in repository https://gitbox.apache.org/repos/asf/artemis.git
commit 8e86dad56775ea6ee04aac2c350408bc91e820be
Author: Justin Bertram <jb...@apache.org>
AuthorDate: Sun Aug 23 15:02:20 2026 -0500
ARTEMIS-6203 make MQTT resiliency soak tests more robust
There is a race in the soak tests which focus on resiliency for MQTT
subscribers. There is a small window in between when the subscriber
receives the initial PUBLISH packet (and records the receipt of the
message) and when the broker receives the PUBREC and acknowledges the
message. It's possible that the test will attempt to verify that the
message is acknowledged during this window and fail.
The fix is simply to use Wait to verify the acknowledgement.
(cherry picked from commit 434da6819f5a27f984f78b37c5bb37029fe563e1)
---
.../main/java/org/apache/activemq/artemis/utils/Wait.java | 8 ++++++--
.../soak/mqtt/resiliency/QoS2CombinedResiliencySoakTest.java | 2 +-
.../mqtt/resiliency/QoS2SubscriberResiliencySoakTest.java | 12 ++++++------
3 files changed, 13 insertions(+), 9 deletions(-)
diff --git a/artemis-unit-test-support/src/main/java/org/apache/activemq/artemis/utils/Wait.java b/artemis-unit-test-support/src/main/java/org/apache/activemq/artemis/utils/Wait.java
index 5c112647ae..f4c1c4191a 100644
--- a/artemis-unit-test-support/src/main/java/org/apache/activemq/artemis/utils/Wait.java
+++ b/artemis-unit-test-support/src/main/java/org/apache/activemq/artemis/utils/Wait.java
@@ -78,7 +78,7 @@ public class Wait {
}
public static void assertEquals(Object obj, ObjectCondition condition) throws Exception {
- assertEquals(obj, condition, MAX_WAIT_MILLIS, SLEEP_MILLIS);
+ assertEquals(obj, condition, MAX_WAIT_MILLIS, SLEEP_MILLIS, null);
}
@@ -117,10 +117,14 @@ public class Wait {
public static void assertEquals(Object obj, ObjectCondition condition, long timeout, long sleepMillis) throws Exception {
+ assertEquals(obj, condition, timeout, sleepMillis, null);
+ }
+
+ public static void assertEquals(Object obj, ObjectCondition condition, long timeout, long sleepMillis, Supplier<String> messageSupplier) throws Exception {
boolean result = waitFor(() -> (obj == condition || (obj != null && obj.equals(condition.getObject()))), timeout, sleepMillis);
if (!result) {
- Assertions.assertEquals(obj, condition.getObject());
+ Assertions.assertEquals(obj, condition.getObject(), messageSupplier);
}
}
diff --git a/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/mqtt/resiliency/QoS2CombinedResiliencySoakTest.java b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/mqtt/resiliency/QoS2CombinedResiliencySoakTest.java
index 99a7c753ce..06945f46d8 100644
--- a/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/mqtt/resiliency/QoS2CombinedResiliencySoakTest.java
+++ b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/mqtt/resiliency/QoS2CombinedResiliencySoakTest.java
@@ -171,7 +171,7 @@ public class QoS2CombinedResiliencySoakTest extends QoS2ResiliencySoakTestSuppor
String clientId = getClientId(subscriber);
assertEquals(0, duplicatesPerSubscriber.get(clientId).size(), "Subscriber " + clientId + " received duplicates: " + duplicatesPerSubscriber.get(clientId));
assertEquals(NUM_PUBLISHERS * NUM_MESSAGES, receivedPerSubscriber.get(clientId).size(), "Subscriber " + clientId + " didn't receive: " + getMissingMessages(publishResult.sentMessages(), receivedPerSubscriber.get(clientId)));
- assertEquals(0L, getSubscriptionQueue(TOPIC, clientId).getMessageCount(), "Subscription queue for " + clientId + " has incorrect message count");
+ Wait.assertEquals(0L, () -> getSubscriptionQueue(TOPIC, clientId).getMessageCount(), 2000, 20, () -> "Subscription queue for " + clientId + " has incorrect message count");
assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(clientId));
assertEquals(0, getSubCacheSize(clientId));
cleanDisconnect(subscriber);
diff --git a/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/mqtt/resiliency/QoS2SubscriberResiliencySoakTest.java b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/mqtt/resiliency/QoS2SubscriberResiliencySoakTest.java
index f456d2a21f..81f546a5ab 100644
--- a/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/mqtt/resiliency/QoS2SubscriberResiliencySoakTest.java
+++ b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/mqtt/resiliency/QoS2SubscriberResiliencySoakTest.java
@@ -172,13 +172,13 @@ public class QoS2SubscriberResiliencySoakTest extends QoS2ResiliencySoakTestSupp
// verify all expected messages received with no duplicates
for (Mqtt5BlockingClient subscriber : subscribers) {
String clientId = getClientId(subscriber);
- assertEquals(0, duplicatesPerSubscriber.get(clientId).size(), "Subscriber " + getClientId(subscriber) + " received duplicates: " + duplicatesPerSubscriber.get(clientId));
- assertEquals(NUM_MESSAGES, receivedPerSubscriber.get(clientId).size(), "Subscriber " + getClientId(subscriber) + " didn't receive: " + getMissingMessages(sentMessages, receivedPerSubscriber.get(clientId)));
- assertEquals(0L, getSubscriptionQueue(TOPIC, getClientId(subscriber)).getMessageCount(), "Subscription queue for " + getClientId(subscriber) + " has incorrect message count");
- assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(getClientId(subscriber)));
- assertEquals(0, getSubCacheSize(getClientId(subscriber)));
+ assertEquals(0, duplicatesPerSubscriber.get(clientId).size(), "Subscriber " + clientId + " received duplicates: " + duplicatesPerSubscriber.get(clientId));
+ assertEquals(NUM_MESSAGES, receivedPerSubscriber.get(clientId).size(), "Subscriber " + clientId + " didn't receive: " + getMissingMessages(sentMessages, receivedPerSubscriber.get(clientId)));
+ Wait.assertEquals(0L, () -> getSubscriptionQueue(TOPIC, clientId).getMessageCount(), 2000, 20, () -> "Subscription queue for " + clientId + " has incorrect message count");
+ assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(clientId));
+ assertEquals(0, getSubCacheSize(clientId));
cleanDisconnect(subscriber);
- assertNull(getSubCache(getClientId(subscriber)), "Sub cache should be null after clean start for " + getClientId(subscriber));
+ assertNull(getSubCache(clientId), "Sub cache should be null after clean start for " + clientId);
}
}
}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@artemis.apache.org
For additional commands, e-mail: commits-help@artemis.apache.org