You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@artemis.apache.org by jb...@apache.org on 2026/08/17 12:42:28 UTC
(artemis) 03/04: ARTEMIS-6189 add QoS 1 & 2 resiliency tests
This is an automated email from the ASF dual-hosted git repository.
jbertram pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/artemis.git
commit 8994ee254e8bec967527fe9d029696e6ebab1998
Author: Justin Bertram <jb...@apache.org>
AuthorDate: Wed Aug 12 14:38:19 2026 -0500
ARTEMIS-6189 add QoS 1 & 2 resiliency tests
Add comprehensive tests for MQTT QoS 1 and QoS 2 resiliency across
broker restarts and client reconnects. These tests use interceptors to
interrupt message flows at every stage of the QoS handshake and verify
correct redelivery, exactly-once semantics, and proper state cleanup.
- QoS 1 subscriber tests cover broker restart and client disconnect
both before and after PUBACK, verifying packet ID reuse with DUP flag
- QoS 1 publisher tests are not needed because the broker has no
intermediate state to maintain; it either persists the inbound message
or it doesn't, and the client drives retransmission entirely on its
own
- QoS 2 publisher tests cover interruptions at all four stages of the
handshake (before/after PUBREC, before/after PUBCOMP) for both broker
restart and client disconnect
- QoS 2 subscriber tests cover the same stages plus session expiry
cleanup via zero-interval disconnect, timed scanner, and clean-start
reconnect
- A soak test runs many publishers and many subscribers sending many
QoS 2 messages with periodic broker restarts, verifying exactly-once
delivery
- HiveMQ MQTT 5 client added as a test dependency
Co-Authored-By: Claude <no...@anthropic.com>
---
artemis-pom/pom.xml | 5 +
etc/checkstyle.xml | 3 +
pom.xml | 1 +
tests/integration-tests/pom.xml | 5 +
.../mqtt5/QoS1SubscriberResiliencyTest.java | 243 ++++++++
.../mqtt5/QoS2PublisherResiliencyTest.java | 535 ++++++++++++++++
.../mqtt5/QoS2SubscriberResiliencyTest.java | 678 +++++++++++++++++++++
.../tests/integration/mqtt5/spec/QoSTests.java | 116 +++-
tests/soak-tests/pom.xml | 6 +-
.../resiliency/QoS2CombinedResiliencySoakTest.java | 188 ++++++
.../QoS2PublisherResiliencySoakTest.java | 153 +++++
.../resiliency/QoS2ResiliencySoakTestSupport.java | 274 +++++++++
.../QoS2SubscriberResiliencySoakTest.java | 184 ++++++
13 files changed, 2383 insertions(+), 8 deletions(-)
diff --git a/artemis-pom/pom.xml b/artemis-pom/pom.xml
index f4d8d7c0b4..003d220496 100644
--- a/artemis-pom/pom.xml
+++ b/artemis-pom/pom.xml
@@ -111,6 +111,11 @@
<artifactId>org.eclipse.paho.mqttv5.client</artifactId>
<version>${paho.client.mqtt.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.hivemq</groupId>
+ <artifactId>hivemq-mqtt-client</artifactId>
+ <version>${hive.client.mqtt.version}</version>
+ </dependency>
<dependency>
<groupId>org.fusesource.mqtt-client</groupId>
<artifactId>mqtt-client</artifactId>
diff --git a/etc/checkstyle.xml b/etc/checkstyle.xml
index de6f3ab67c..2063e4ce60 100644
--- a/etc/checkstyle.xml
+++ b/etc/checkstyle.xml
@@ -88,6 +88,9 @@ under the License.
<module name="NoWhitespaceBefore"/>
<!-- Checks that most tokens are surrounded by whitespace. RCURLY is omitted so we can do stuff like "new Thread(){...}.start()". -->
<module name="WhitespaceAround">
+ <!-- Allows empty record, class, interface, and enum bodies: {} -->
+ <property name="allowEmptyTypes" value="true"/>
+
<property name="tokens" value="ASSIGN, BAND, BAND_ASSIGN, BOR, BOR_ASSIGN, BSR, BSR_ASSIGN, BXOR, BXOR_ASSIGN, COLON, DIV, DIV_ASSIGN, EQUAL, GE, GT, LAND, LCURLY, LE, LITERAL_ASSERT, LITERAL_CATCH, LITERAL_DO, LITERAL_ELSE, LITERAL_FINALLY, LITERAL_FOR, LITERAL_IF, LITERAL_RETURN, LITERAL_SYNCHRONIZED, LITERAL_TRY, LITERAL_WHILE, LOR, LT, MINUS, MINUS_ASSIGN, MOD, MOD_ASSIGN, NOT_EQUAL, PLUS, PLUS_ASSIGN, QUESTION, SL, SLIST, SL_ASSIGN, SR, SR_ASSIGN, STAR, STAR_ASSIGN, TYPE_EX [...]
</module>
<module name="WhitespaceAfter">
diff --git a/pom.xml b/pom.xml
index c6e435b891..23ad397c4f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -174,6 +174,7 @@
<junit4.version>4.13.2</junit4.version>
<version.jaxb.runtime>2.3.9</version.jaxb.runtime>
<paho.client.mqtt.version>1.2.5</paho.client.mqtt.version>
+ <hive.client.mqtt.version>1.3.17</hive.client.mqtt.version>
<postgresql.version>42.7.13</postgresql.version>
<testcontainers.version>1.21.4</testcontainers.version>
<selenium.version>4.46.0</selenium.version>
diff --git a/tests/integration-tests/pom.xml b/tests/integration-tests/pom.xml
index fddff750a4..7987103b4a 100644
--- a/tests/integration-tests/pom.xml
+++ b/tests/integration-tests/pom.xml
@@ -195,6 +195,11 @@
<artifactId>org.eclipse.paho.mqttv5.client</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>com.hivemq</groupId>
+ <artifactId>hivemq-mqtt-client</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>jakarta.resource</groupId>
<artifactId>jakarta.resource-api</artifactId>
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/QoS1SubscriberResiliencyTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/QoS1SubscriberResiliencyTest.java
new file mode 100644
index 0000000000..e4eaa71e24
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/QoS1SubscriberResiliencyTest.java
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.integration.mqtt5;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.netty.handler.codec.mqtt.MqttMessageType;
+import io.netty.handler.codec.mqtt.MqttPublishMessage;
+import org.apache.activemq.artemis.core.protocol.mqtt.MQTTInterceptor;
+import org.apache.activemq.artemis.utils.RandomUtil;
+import org.apache.activemq.artemis.utils.Wait;
+import org.eclipse.paho.mqttv5.client.MqttClient;
+import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
+import org.eclipse.paho.mqttv5.client.MqttConnectionOptionsBuilder;
+import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for QoS 1 protocol resiliency with subscriber reconnections or broker restarts.
+ * <p>
+ * QoS 1 Protocol Flow (broker sending to subscriber):
+ * <ol>
+ * <li>Broker sends PUBLISH (QoS=1)</li>
+ * <li>Subscriber sends PUBACK</li>
+ * </ol>
+ * These tests verify that the protocol maintains at-least-once delivery semantics when either clients
+ * reconnect or the broker is restarted at each stage of the QoS 1 flow.
+ */
+public class QoS1SubscriberResiliencyTest extends MQTT5TestSupport {
+
+ protected static final long DEFAULT_TIMEOUT_SEC = 10;
+
+ @Override
+ public boolean isProtocolLoggingEnabled() {
+ return true;
+ }
+
+ /**
+ * Verifies that the broker will re-use the same packet ID if it sends a PUBLISH but fails to receive the
+ * corresponding PUBACK.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS1BrokerRestartBeforePubAckSent() throws Exception {
+ testQoS1FailureBeforePubAckSent(true);
+ }
+
+ /**
+ * Same test as {@link testQoS1BrokerRestartBeforePubAckSent} but disconnecting the client instead of restarting the
+ * broker.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS1ClientDisconnectBeforePubAckSent() throws Exception {
+ testQoS1FailureBeforePubAckSent(false);
+ }
+
+ public void testQoS1FailureBeforePubAckSent(boolean restart) throws Exception {
+ final String TOPIC = "test/resiliency";
+ final String SUBSCRIBER_CLIENT_ID = "subscriber";
+ final String PUBLISHER_CLIENT_ID = "publisher";
+
+ // Set up interceptor to block the *second* incoming PUBACK.
+ // We allow 1 PUBACK so the packet ID goes up to 2 for testing.
+ final CountDownLatch pubAckLatch = new CountDownLatch(1);
+ AtomicInteger pubAckCount = new AtomicInteger(0);
+ final CountDownLatch stopLatch = new CountDownLatch(1);
+ MQTTInterceptor pubAckInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBACK && pubAckCount.incrementAndGet() > 1) {
+ pubAckLatch.countDown();
+ logger.info("Blocking incoming {}", packet.fixedHeader().messageType());
+ try {
+ stopLatch.await(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ return false;
+ }
+ logger.info("Allowing incoming {}", packet.fixedHeader().messageType());
+ return true;
+ };
+ server.getRemotingService().addIncomingInterceptor(pubAckInterceptor);
+
+ // Consumer with persistent session
+ MqttClient subscriber = createPahoClient(SUBSCRIBER_CLIENT_ID);
+ subscriber.setCallback(new DefaultMqttCallback() {
+ @Override
+ public void messageArrived(String topic, MqttMessage message) throws Exception {
+ logger.info("messageArrived({}, {})", topic, message);
+ }
+ });
+ MqttConnectionOptions subscriberOptions = new MqttConnectionOptionsBuilder()
+ .cleanStart(false)
+ .sessionExpiryInterval(300L)
+ .build();
+ subscriber.connect(subscriberOptions);
+ subscriber.subscribe(TOPIC, 1);
+
+ // Producer
+ MqttClient producer = createPahoClient(PUBLISHER_CLIENT_ID);
+ producer.connect();
+
+ // Send 2 messages to ensure the packet ID is preserved by the broker.
+ // If we just send 1 message it won't be clear if the broker just started generating packet IDs from scratch.
+ producer.publish(TOPIC, RandomUtil.randomBytes(), 1, false);
+ producer.publish(TOPIC, RandomUtil.randomBytes(), 1, false);
+
+ producer.disconnect();
+ producer.close();
+
+ assertTrue(pubAckLatch.await(5, TimeUnit.SECONDS));
+
+ stopLatch.countDown();
+ if (restart) {
+ server.stop();
+ waitForServerToStop(server);
+ server.start();
+ waitForServerToStart(server);
+ } else {
+ server.getRemotingService().clearInterceptors();
+ server.getActiveMQServerControl().closeConnectionWithID(server.getActiveMQServerControl().listConnectionIDs()[0]);
+ }
+
+ assertTrue(getProtocolManager().getStateManager().packetIdCorrelationExists(SUBSCRIBER_CLIENT_ID, 2));
+
+ CountDownLatch packetIdLatch = new CountDownLatch(1);
+ MQTTInterceptor pubInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBLISH) {
+ if (((MqttPublishMessage)packet).variableHeader().packetId() == 2 && ((MqttPublishMessage)packet).fixedHeader().isDup()) {
+ packetIdLatch.countDown();
+ }
+ }
+ return true;
+ };
+ server.getRemotingService().addOutgoingInterceptor(pubInterceptor);
+
+ reconnectSafely(subscriber);
+
+ assertTrue(packetIdLatch.await(5, TimeUnit.SECONDS), "Didn't find a duplicate PUBLISH with the expected packet id");
+
+ Wait.assertEquals(0L, () -> getSubscriptionQueue(TOPIC, SUBSCRIBER_CLIENT_ID).getMessageCount(), 500, 25);
+ assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(SUBSCRIBER_CLIENT_ID));
+
+ subscriber.disconnect();
+ subscriber.close();
+ }
+
+ /**
+ * Verifies that after a complete QoS 1 protocol exchange (PUBLISH, PUBACK), a broker restart does
+ * not cause re-delivery.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS1BrokerRestartAfterPubAckSent() throws Exception {
+ testQoS1FailureAfterPubAckSent(true);
+ }
+
+ /**
+ * Same test as {@link testQoS1BrokerRestartAfterPubAckSent} but disconnecting the client instead of restarting the
+ * broker.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS1ClientDisconnectAfterPubAckSent() throws Exception {
+ testQoS1FailureAfterPubAckSent(false);
+ }
+
+ public void testQoS1FailureAfterPubAckSent(boolean restart) throws Exception {
+ final String TOPIC = "test/resiliency";
+ final String SUBSCRIBER_CLIENT_ID = "subscriber";
+ final String PUBLISHER_CLIENT_ID = "publisher";
+ AtomicInteger messageCount = new AtomicInteger(0);
+
+ // Subscriber with persistent session
+ MqttClient subscriber = createPahoClient(SUBSCRIBER_CLIENT_ID);
+ subscriber.setCallback(new DefaultMqttCallback() {
+ @Override
+ public void messageArrived(String topic, MqttMessage message) throws Exception {
+ messageCount.incrementAndGet();
+ logger.info("messageArrived({}, {})", topic, message);
+ }
+ });
+ MqttConnectionOptions subscriberOptions = new MqttConnectionOptionsBuilder()
+ .cleanStart(false)
+ .sessionExpiryInterval(300L)
+ .build();
+ subscriber.connect(subscriberOptions);
+ subscriber.subscribe(TOPIC, 1);
+
+ MqttClient producer = createPahoClient(PUBLISHER_CLIENT_ID);
+ producer.connect();
+ producer.publish(TOPIC, RandomUtil.randomBytes(), 1, false);
+ producer.disconnect();
+ producer.close();
+
+ Wait.assertEquals(1L, () -> messageCount.get(), 500, 25);
+ Wait.assertEquals(1L, () -> getSubscriptionQueue(TOPIC, SUBSCRIBER_CLIENT_ID).getMessagesAcknowledged(), 500, 25);
+ Wait.assertEquals(0L, () -> getSubscriptionQueue(TOPIC, SUBSCRIBER_CLIENT_ID).getMessageCount(), 500, 25);
+
+ assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(SUBSCRIBER_CLIENT_ID));
+
+ if (restart) {
+ server.stop();
+ waitForServerToStop(server);
+ server.start();
+ waitForServerToStart(server);
+ } else {
+ server.getRemotingService().clearInterceptors();
+ server.getActiveMQServerControl().closeConnectionWithID(server.getActiveMQServerControl().listConnectionIDs()[0]);
+ }
+
+ reconnectSafely(subscriber);
+
+ // Check for any unexpected re-delivery
+ assertFalse(Wait.waitFor(() -> messageCount.get() > 1, 500, 25));
+
+ assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(SUBSCRIBER_CLIENT_ID));
+
+ subscriber.disconnect();
+ subscriber.close();
+ }
+}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/QoS2PublisherResiliencyTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/QoS2PublisherResiliencyTest.java
new file mode 100644
index 0000000000..66d09dcd21
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/QoS2PublisherResiliencyTest.java
@@ -0,0 +1,535 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.integration.mqtt5;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import io.netty.handler.codec.mqtt.MqttMessageType;
+import org.apache.activemq.artemis.api.core.QueueConfiguration;
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.core.protocol.mqtt.MQTTInterceptor;
+import org.apache.activemq.artemis.utils.RandomUtil;
+import org.apache.activemq.artemis.utils.ReusableLatch;
+import org.apache.activemq.artemis.utils.Wait;
+import org.eclipse.paho.mqttv5.client.MqttClient;
+import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
+import org.eclipse.paho.mqttv5.client.MqttConnectionOptionsBuilder;
+import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse;
+import org.eclipse.paho.mqttv5.common.MqttException;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for QoS 2 protocol resiliency with publisher reconnections or broker restarts.
+ * <p>
+ * QoS 2 Protocol Flow (publisher sending to broker):
+ * <ol>
+ * <li>Publisher sends PUBLISH (QoS=2)</li>
+ * <li>Broker sends PUBREC</li>
+ * <li>Publisher sends PUBREL</li>
+ * <li>Broker sends PUBCOMP</li>
+ * </ol>
+ * These tests verify that the protocol maintains exactly-once delivery semantics when either clients
+ * reconnect or the broker is restarted at each stage of the QoS 2 flow.
+ */
+public class QoS2PublisherResiliencyTest extends MQTT5TestSupport {
+
+ protected static final long DEFAULT_TIMEOUT_SEC = 10;
+
+ @Override
+ public boolean isProtocolLoggingEnabled() {
+ return true;
+ }
+
+ /**
+ * Verifies that resending the same message via QoS2 doesn't result in duplicates when the broker is restarted after
+ * it receives the PUBLISH but before it sends the PUBREC. In this circumstance the client will reconnect and send
+ * the PUBLISH again with the same packet ID and the dup flag set to true.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2BrokerRestartBeforePubRecSent() throws Exception {
+ testQoS2FailureBeforePubRecSent(true);
+ }
+
+ /**
+ * Same test as {@link testQoS2BrokerRestartBeforePubRecSent} but disconnecting the client instead of restarting the
+ * broker.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2ClientDisconnectBeforePubRecSent() throws Exception {
+ testQoS2FailureBeforePubRecSent(false);
+ }
+
+ public void testQoS2FailureBeforePubRecSent(boolean restart) throws Exception {
+ final String TOPIC = RandomUtil.randomUUIDString();
+ final String CLIENTID = "publisher";
+ final CountDownLatch publishLatch = new CountDownLatch(1);
+ final CountDownLatch stopLatch = new CountDownLatch(1);
+ final CountDownLatch pubCompLatch = new CountDownLatch(1);
+
+ // Simulate a subscription queue
+ server.createQueue(QueueConfiguration.of(TOPIC)
+ .setAddress(TOPIC)
+ .setRoutingType(RoutingType.MULTICAST)
+ .setDurable(true));
+
+ // Set up interceptor block the initial PUBLISH
+ MQTTInterceptor pubRecInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBLISH) {
+ publishLatch.countDown();
+ logger.info("Blocking incoming {}", packet.fixedHeader().messageType());
+ try {
+ stopLatch.await(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ return false;
+ }
+ logger.info("Allowing incoming {}", packet.fixedHeader().messageType());
+ return true;
+ };
+ server.getRemotingService().addIncomingInterceptor(pubRecInterceptor);
+
+ // Producer with persistent session
+ MqttClient publisher = createPahoClient(CLIENTID);
+ publisher.setCallback(new DefaultMqttCallback() {
+ @Override
+ public void disconnected(MqttDisconnectResponse disconnectResponse) {
+ logger.info("{} disconnected", CLIENTID);
+ }
+ });
+ MqttConnectionOptions producerOptions = new MqttConnectionOptionsBuilder()
+ .cleanStart(false)
+ .sessionExpiryInterval(300L)
+ .build();
+ publisher.connect(producerOptions);
+
+ assertNull(getPubCache(CLIENTID));
+
+ // Send message async as it will block waiting for a PUBREC that won't come
+ CompletableFuture.runAsync(() -> {
+ try {
+ publisher.publish(TOPIC, RandomUtil.randomBytes(), EXACTLY_ONCE, false);
+ } catch (MqttException e) {
+ logger.info(e.getMessage());
+ }
+ });
+
+ assertTrue(publishLatch.await(5, TimeUnit.SECONDS));
+ stopLatch.countDown();
+ if (restart) {
+ server.stop();
+ waitForServerToStop(server);
+ server.start();
+ waitForServerToStart(server);
+ } else {
+ server.getRemotingService().clearInterceptors();
+ server.getActiveMQServerControl().closeConnectionWithID(server.getActiveMQServerControl().listConnectionIDs()[0]);
+ }
+
+ MQTTInterceptor pubCompInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBCOMP) {
+ pubCompLatch.countDown();
+ }
+ return true;
+ };
+ server.getRemotingService().addOutgoingInterceptor(pubCompInterceptor);
+
+ assertNull(getPubCache(CLIENTID));
+
+ // The client will automatically re-initiate the QoS2 protocol after reconnecting since it never got a PUBREC
+ reconnectSafely(publisher);
+
+ // Wait for the PUBCOMP to confirm QoS2 protocol is done
+ assertTrue(pubCompLatch.await(5, TimeUnit.SECONDS));
+
+ // Verify only one message is in the queue despite the QoS2 interruption
+ Wait.assertEquals(1L, () -> server.locateQueue(TOPIC).getMessageCount(), 500, 25);
+
+ publisher.disconnect();
+
+ assertEquals(0, getPubCacheSize(CLIENTID));
+
+ // connect again to clean the session which will completely remove the cache from memory and disk
+ publisher.connect(new MqttConnectionOptionsBuilder().cleanStart(true).sessionExpiryInterval(0L).build());
+ assertNull(getPubCache(CLIENTID));
+ publisher.disconnect();
+ publisher.close();
+ }
+
+ /**
+ * Verifies that resending the same message via QoS2 doesn't result in duplicates when the broker is restarted after
+ * it receives the PUBLISH and sends the PUBREC but before the client receives the PUBREC. In this circumstance the
+ * client will reconnect and send the PUBLISH again with the same packet ID and the dup flag set to true.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2BrokerRestartAfterPubRecSent() throws Exception {
+ testQoS2FailureAfterPubRecSent(true);
+ }
+
+ /**
+ * Same test as {@link testQoS2BrokerRestartAfterPubRecSent} but disconnecting the client instead of restarting the
+ * broker.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2ClientDisconnectAfterPubRecSent() throws Exception {
+ testQoS2FailureAfterPubRecSent(false);
+ }
+
+ public void testQoS2FailureAfterPubRecSent(boolean restart) throws Exception {
+ final String TOPIC = RandomUtil.randomUUIDString();
+ final String CLIENTID = "publisher";
+ final CountDownLatch pubRecLatch = new CountDownLatch(1);
+ final CountDownLatch pubCompLatch = new CountDownLatch(1);
+
+ // Simulate a subscription queue
+ server.createQueue(QueueConfiguration.of(TOPIC)
+ .setAddress(TOPIC)
+ .setRoutingType(RoutingType.MULTICAST)
+ .setDurable(true));
+
+ // Set up interceptor block the initial PUBREC
+ MQTTInterceptor pubRecInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBREC) {
+ pubRecLatch.countDown();
+ logger.info("Blocking outgoing {}", packet.fixedHeader().messageType());
+ return false;
+ }
+
+ logger.info("Allowing outgoing {}", packet.fixedHeader().messageType());
+ return true;
+ };
+ server.getRemotingService().addOutgoingInterceptor(pubRecInterceptor);
+
+ // Producer with persistent session
+ MqttClient publisher = createPahoClient(CLIENTID);
+ publisher.setCallback(new DefaultMqttCallback() {
+ @Override
+ public void disconnected(MqttDisconnectResponse disconnectResponse) {
+ logger.info("{} disconnected", CLIENTID);
+ }
+ });
+ MqttConnectionOptions producerOptions = new MqttConnectionOptionsBuilder()
+ .cleanStart(false)
+ .sessionExpiryInterval(300L)
+ .build();
+ publisher.connect(producerOptions);
+
+ assertEquals(0, server.locateQueue(TOPIC).getMessageCount());
+ assertNull(getPubCache(CLIENTID));
+
+ // Send message async as it will block waiting for a PUBREC that won't come
+ CompletableFuture.runAsync(() -> {
+ try {
+ publisher.publish(TOPIC, RandomUtil.randomBytes(), EXACTLY_ONCE, false);
+ } catch (MqttException e) {
+ logger.info(e.getMessage());
+ }
+ });
+ assertTrue(pubRecLatch.await(5, TimeUnit.SECONDS));
+
+ if (restart) {
+ server.stop();
+ waitForServerToStop(server);
+ server.start();
+ waitForServerToStart(server);
+ } else {
+ server.getRemotingService().clearInterceptors();
+ server.getActiveMQServerControl().closeConnectionWithID(server.getActiveMQServerControl().listConnectionIDs()[0]);
+ }
+
+ MQTTInterceptor pubCompInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBCOMP) {
+ pubCompLatch.countDown();
+ }
+ return true;
+ };
+ server.getRemotingService().addOutgoingInterceptor(pubCompInterceptor);
+
+ assertEquals(1, getPubCacheSize(CLIENTID));
+
+ // The client will automatically re-initiate the QoS2 protocol after reconnecting since it never got a PUBREC
+ reconnectSafely(publisher);
+
+ // Wait for the PUBCOMP to confirm QoS2 protocol is done
+ assertTrue(pubCompLatch.await(5, TimeUnit.SECONDS));
+
+ // Verify only one message is in the queue despite the QoS2 interruption
+ Wait.assertEquals(1L, () -> server.locateQueue(TOPIC).getMessageCount(), 500, 25);
+
+ publisher.disconnect();
+
+ assertEquals(0, getPubCacheSize(CLIENTID));
+
+ // connect again to clean the session which will completely remove the cache from memory and disk
+ publisher.connect(new MqttConnectionOptionsBuilder().cleanStart(true).sessionExpiryInterval(0L).build());
+ assertNull(getPubCache(CLIENTID));
+ publisher.disconnect();
+ publisher.close();
+ }
+
+ /**
+ * Verifies that resending the same message via QoS2 doesn't result in duplicates when the broker is restarted after
+ * it receives the PUBREL but before it sends the PUBCOMP. In this circumstance the client will reconnect and send
+ * the PUBREL again with the same packet ID.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2BrokerRestartBeforePubCompSent() throws Exception {
+ testQoS2FailureBeforePubCompSent(true);
+ }
+
+ /**
+ * Same test as {@link testQoS2BrokerRestartBeforePubCompSent} but disconnecting the client instead of restarting the
+ * broker.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2ClientDisconnectBeforePubCompSent() throws Exception {
+ testQoS2FailureBeforePubCompSent(false);
+ }
+
+ public void testQoS2FailureBeforePubCompSent(boolean restart) throws Exception {
+ final String TOPIC = RandomUtil.randomUUIDString();
+ final String CLIENTID = "publisher";
+ final CountDownLatch pubRelLatch = new CountDownLatch(1);
+ final CountDownLatch stopLatch = new CountDownLatch(1);
+ final CountDownLatch pubCompLatch = new CountDownLatch(1);
+
+ // Simulate a subscription queue
+ server.createQueue(QueueConfiguration.of(TOPIC)
+ .setAddress(TOPIC)
+ .setRoutingType(RoutingType.MULTICAST)
+ .setDurable(true));
+
+ // Set up interceptor block the initial PUBREL
+ MQTTInterceptor pubRelInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBREL) {
+ pubRelLatch.countDown();
+ logger.info("Blocking incoming {}", packet.fixedHeader().messageType());
+ try {
+ assertTrue(stopLatch.await(5, TimeUnit.SECONDS));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ return false;
+ }
+ logger.info("Allowing incoming {}", packet.fixedHeader().messageType());
+ return true;
+ };
+ server.getRemotingService().addIncomingInterceptor(pubRelInterceptor);
+
+ // Producer with persistent session
+ MqttClient publisher = createPahoClient(CLIENTID);
+ publisher.setCallback(new DefaultMqttCallback() {
+ @Override
+ public void disconnected(MqttDisconnectResponse disconnectResponse) {
+ logger.info("{} disconnected", CLIENTID);
+ }
+ });
+ MqttConnectionOptions producerOptions = new MqttConnectionOptionsBuilder()
+ .cleanStart(false)
+ .sessionExpiryInterval(300L)
+ .build();
+ publisher.connect(producerOptions);
+
+ assertNull(getPubCache(CLIENTID));
+
+ // Send message async as it will block waiting for a PUBCOMP that won't come
+ CompletableFuture.runAsync(() -> {
+ try {
+ publisher.publish(TOPIC, RandomUtil.randomBytes(), EXACTLY_ONCE, false);
+ } catch (MqttException e) {
+ logger.info(e.getMessage());
+ }
+ });
+ assertTrue(pubRelLatch.await(5, TimeUnit.SECONDS));
+
+ stopLatch.countDown();
+
+ if (restart) {
+ server.stop();
+ waitForServerToStop(server);
+ server.start();
+ waitForServerToStart(server);
+ } else {
+ server.getRemotingService().clearInterceptors();
+ server.getActiveMQServerControl().closeConnectionWithID(server.getActiveMQServerControl().listConnectionIDs()[0]);
+ }
+
+ MQTTInterceptor pubCompInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBCOMP) {
+ pubCompLatch.countDown();
+ }
+ return true;
+ };
+ server.getRemotingService().addOutgoingInterceptor(pubCompInterceptor);
+
+ assertEquals(1, getPubCacheSize(CLIENTID));
+
+ // The client will automatically re-initiate the QoS2 protocol after reconnecting since it never got a PUBCOMP
+ reconnectSafely(publisher);
+
+ // Wait for the PUBCOMP to confirm QoS2 protocol is done
+ assertTrue(pubCompLatch.await(5, TimeUnit.SECONDS));
+
+ // Verify only one message is in the queue despite the QoS2 interruption
+ Wait.assertEquals(1L, () -> server.locateQueue(TOPIC).getMessageCount(), 500, 25);
+
+ publisher.disconnect();
+
+ assertEquals(0, getPubCacheSize(CLIENTID));
+
+ // connect again to clean the session which will completely remove the cache from memory and disk
+ publisher.connect(new MqttConnectionOptionsBuilder().cleanStart(true).sessionExpiryInterval(0L).build());
+ assertNull(getPubCache(CLIENTID));
+ publisher.disconnect();
+ publisher.close();
+ }
+
+ /**
+ * Verifies that resending the same message via QoS2 doesn't result in duplicates when the broker is restarted after
+ * it receives the PUBREL and sends the PUBCOMP but before the client receives the PUBCOMP. In this circumstance the
+ * client will reconnect and send the PUBREL again with the same packet ID.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2BrokerRestartAfterPubCompSent() throws Exception {
+ testQoS2FailureAfterPubCompSent(true);
+ }
+
+ /**
+ * Same test as {@link testQoS2BrokerRestartAfterPubCompSent} but disconnecting the client instead of restarting the
+ * broker.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2ClientDisconnectAfterPubCompSent() throws Exception {
+ testQoS2FailureAfterPubCompSent(false);
+ }
+
+ public void testQoS2FailureAfterPubCompSent(boolean restart) throws Exception {
+ final String TOPIC = RandomUtil.randomUUIDString();
+ final String CLIENTID = "publisher";
+ final ReusableLatch pubCompLatch = new ReusableLatch(1);
+
+ // Simulate a subscription queue
+ server.createQueue(QueueConfiguration.of(TOPIC)
+ .setAddress(TOPIC)
+ .setRoutingType(RoutingType.MULTICAST)
+ .setDurable(true));
+
+ // Set up interceptor block the initial PUBCOMP
+ MQTTInterceptor initialPubCompInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBCOMP) {
+ pubCompLatch.countDown();
+ logger.info("Blocking outgoing {}", packet.fixedHeader().messageType());
+ return false;
+ }
+
+ logger.info("Allowing outgoing {}", packet.fixedHeader().messageType());
+ return true;
+ };
+ server.getRemotingService().addOutgoingInterceptor(initialPubCompInterceptor);
+
+ // Producer with persistent session
+ MqttClient publisher = createPahoClient(CLIENTID);
+ publisher.setCallback(new DefaultMqttCallback() {
+ @Override
+ public void disconnected(MqttDisconnectResponse disconnectResponse) {
+ logger.info("{} disconnected", CLIENTID);
+ }
+ });
+ MqttConnectionOptions producerOptions = new MqttConnectionOptionsBuilder()
+ .cleanStart(false)
+ .sessionExpiryInterval(300L)
+ .build();
+ publisher.connect(producerOptions);
+
+ assertEquals(0, server.locateQueue(TOPIC).getMessageCount());
+ assertNull(getPubCache(CLIENTID));
+
+ // Send message async as it will block waiting for a PUBREC that won't come
+ CompletableFuture.runAsync(() -> {
+ try {
+ publisher.publish(TOPIC, RandomUtil.randomBytes(), EXACTLY_ONCE, false);
+ } catch (MqttException e) {
+ logger.info(e.getMessage());
+ }
+ });
+ assertTrue(pubCompLatch.await(5, TimeUnit.SECONDS));
+
+ if (restart) {
+ server.stop();
+ waitForServerToStop(server);
+ server.start();
+ waitForServerToStart(server);
+ } else {
+ server.getRemotingService().clearInterceptors();
+ server.getActiveMQServerControl().closeConnectionWithID(server.getActiveMQServerControl().listConnectionIDs()[0]);
+ }
+
+ pubCompLatch.countUp();
+ MQTTInterceptor pubCompInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBCOMP) {
+ pubCompLatch.countDown();
+ }
+ return true;
+ };
+ server.getRemotingService().addOutgoingInterceptor(pubCompInterceptor);
+
+ if (restart) {
+ assertNull(getPubCache(CLIENTID));
+ } else {
+ assertEquals(0, getPubCacheSize(CLIENTID));
+ }
+
+ // The client will automatically re-initiate the QoS2 protocol after reconnecting since it never got a PUBCOMP
+ reconnectSafely(publisher);
+
+ // Wait for the PUBCOMP to confirm QoS2 protocol is done
+ assertTrue(pubCompLatch.await(5, TimeUnit.SECONDS));
+
+ // Verify only one message is in the queue despite the QoS2 interruption
+ Wait.assertEquals(1L, () -> server.locateQueue(TOPIC).getMessageCount(), 500, 25);
+
+ publisher.disconnect();
+
+ if (restart) {
+ assertNull(getPubCache(CLIENTID));
+ publisher.close();
+ } else {
+ assertEquals(0, getPubCacheSize(CLIENTID));
+
+ // connect again to clean the session which will completely remove the cache from memory and disk
+ publisher.connect(new MqttConnectionOptionsBuilder().cleanStart(true).sessionExpiryInterval(0L).build());
+ assertNull(getPubCache(CLIENTID));
+ publisher.disconnect();
+ publisher.close();
+ }
+ }
+}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/QoS2SubscriberResiliencyTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/QoS2SubscriberResiliencyTest.java
new file mode 100644
index 0000000000..353449100a
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/QoS2SubscriberResiliencyTest.java
@@ -0,0 +1,678 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.integration.mqtt5;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.netty.handler.codec.mqtt.MqttMessageType;
+import io.netty.handler.codec.mqtt.MqttPublishMessage;
+import org.apache.activemq.artemis.core.protocol.mqtt.MQTTInterceptor;
+import org.apache.activemq.artemis.utils.ByteUtil;
+import org.apache.activemq.artemis.utils.RandomUtil;
+import org.apache.activemq.artemis.utils.Wait;
+import org.eclipse.paho.mqttv5.client.MqttClient;
+import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
+import org.eclipse.paho.mqttv5.client.MqttConnectionOptionsBuilder;
+import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for QoS 2 protocol resiliency with subscriber reconnections or broker restarts.
+ * <p>
+ * QoS 2 Protocol Flow (broker sending to subscriber):
+ * <ol>
+ * <li>Broker sends PUBLISH (QoS=2)</li>
+ * <li>Subscriber sends PUBREC</li>
+ * <li>Broker sends PUBREL</li>
+ * <li>Subscriber sends PUBCOMP</li>
+ * </ol>
+ * These tests verify that the protocol maintains exactly-once delivery semantics when either clients
+ * reconnect or the broker is restarted at each stage of the QoS 2 flow.
+ */
+public class QoS2SubscriberResiliencyTest extends MQTT5TestSupport {
+
+ protected static final long DEFAULT_TIMEOUT_SEC = 10;
+
+ @Override
+ public boolean isProtocolLoggingEnabled() {
+ return true;
+ }
+
+ /**
+ * Verifies that the broker will re-use the same packet ID if it sends a PUBLISH but fails to receive the
+ * corresponding PUBREC.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2BrokerRestartBeforePubRecSent() throws Exception {
+ testQoS2FailureBeforePubRecSent(true);
+ }
+
+ /**
+ * Same test as {@link testQoS2BrokerRestartBeforePubRecSent} but disconnecting the client instead of restarting the
+ * broker.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2ClientDisconnectBeforePubRecSent() throws Exception {
+ testQoS2FailureBeforePubRecSent(false);
+ }
+
+ public void testQoS2FailureBeforePubRecSent(boolean restart) throws Exception {
+ final String TOPIC = "test/resiliency";
+ final String SUBSCRIBER_CLIENT_ID = "subscriber";
+ final String PUBLISHER_CLIENT_ID = "publisher";
+
+ // Set up interceptor to block the *second* incoming PUBREC.
+ // Allow 1 PUBREC so the packet ID goes up to 2 for testing.
+ final CountDownLatch pubRecLatch = new CountDownLatch(1);
+ AtomicInteger pubRecCount = new AtomicInteger(0);
+ final CountDownLatch stopLatch = new CountDownLatch(1);
+ MQTTInterceptor pubRecInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBREC && pubRecCount.incrementAndGet() > 1) {
+ pubRecLatch.countDown();
+ logger.info("Blocking incoming {}", packet.fixedHeader().messageType());
+ try {
+ stopLatch.await(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ return false;
+ }
+ logger.info("Allowing incoming {}", packet.fixedHeader().messageType());
+ return true;
+ };
+ server.getRemotingService().addIncomingInterceptor(pubRecInterceptor);
+
+ // Consumer with persistent session
+ MqttClient subscriber = createPahoClient(SUBSCRIBER_CLIENT_ID);
+ subscriber.setCallback(new DefaultMqttCallback() {
+ @Override
+ public void messageArrived(String topic, MqttMessage message) throws Exception {
+ logger.info("messageArrived({}, {})", topic, message);
+ }
+ });
+ MqttConnectionOptions subscriberOptions = new MqttConnectionOptionsBuilder()
+ .cleanStart(false)
+ .sessionExpiryInterval(300L)
+ .build();
+ subscriber.connect(subscriberOptions);
+ subscriber.subscribe(TOPIC, EXACTLY_ONCE);
+
+ // Producer
+ MqttClient producer = createPahoClient(PUBLISHER_CLIENT_ID);
+ producer.connect();
+
+ // Send 2 messages to ensure the packet ID is preserved by the broker.
+ // If we just send 1 message it won't be clear if the broker just started generating packet IDs from scratch.
+ producer.publish(TOPIC, RandomUtil.randomBytes(), EXACTLY_ONCE, false);
+ producer.publish(TOPIC, RandomUtil.randomBytes(), EXACTLY_ONCE, false);
+
+ Wait.assertEquals(0, () -> getPubCacheSize(PUBLISHER_CLIENT_ID));
+
+ producer.disconnect();
+ producer.close();
+
+ assertNull(getPubCache(PUBLISHER_CLIENT_ID));
+
+ assertTrue(pubRecLatch.await(5, TimeUnit.SECONDS));
+ stopLatch.countDown();
+
+ if (restart) {
+ server.stop();
+ waitForServerToStop(server);
+ server.start();
+ waitForServerToStart(server);
+ } else {
+ server.getRemotingService().clearInterceptors();
+ Wait.assertEquals(0, () -> server.getRemotingService().getConnections().size(), 1000, 10);
+ }
+
+ assertTrue(getProtocolManager().getStateManager().packetIdCorrelationExists(SUBSCRIBER_CLIENT_ID, 2));
+
+ final CountDownLatch pubCompLatch = new CountDownLatch(1);
+ MQTTInterceptor pubCompInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBCOMP) {
+ pubCompLatch.countDown();
+ }
+ return true;
+ };
+ server.getRemotingService().addIncomingInterceptor(pubCompInterceptor);
+
+ CountDownLatch packetIdLatch = new CountDownLatch(1);
+ MQTTInterceptor pubInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBLISH) {
+ if (((MqttPublishMessage)packet).variableHeader().packetId() == 2 && ((MqttPublishMessage)packet).fixedHeader().isDup()) {
+ packetIdLatch.countDown();
+ }
+ }
+ return true;
+ };
+ server.getRemotingService().addOutgoingInterceptor(pubInterceptor);
+
+ reconnectSafely(subscriber);
+
+ assertTrue(packetIdLatch.await(5, TimeUnit.SECONDS), "Didn't find a duplicate PUBLISH with the expected packet id");
+ assertTrue(pubCompLatch.await(5, TimeUnit.SECONDS));
+
+ Wait.assertEquals(0L, () -> getSubscriptionQueue(TOPIC, SUBSCRIBER_CLIENT_ID).getMessageCount(), 500, 25);
+
+ assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(SUBSCRIBER_CLIENT_ID));
+ Wait.assertEquals(0, () -> getPubRecCacheSize(SUBSCRIBER_CLIENT_ID));
+
+ subscriber.disconnect();
+
+ // connect again to clean the session which will completely remove the cache from memory and disk
+ subscriber.connect(new MqttConnectionOptionsBuilder().cleanStart(true).sessionExpiryInterval(0L).build());
+ assertNull(getPubRecCache(SUBSCRIBER_CLIENT_ID));
+ subscriber.disconnect();
+ subscriber.close();
+ }
+
+ /**
+ * Verifies that the broker correctly completes the QoS 2 flow if it receives the PUBREC but the corresponding
+ * PUBREL is not sent to the consumer before the broker restarts.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2BrokerRestartAfterPubRecSent() throws Exception {
+ testQoS2FailureAfterPubRecSent(true);
+ }
+
+ /**
+ * Same test as {@link testQoS2BrokerRestartAfterPubRecSent} but disconnecting the client instead of restarting the
+ * broker.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2ClientDisconnectAfterPubRecSent() throws Exception {
+ testQoS2FailureAfterPubRecSent(false);
+ }
+
+ public void testQoS2FailureAfterPubRecSent(boolean restart) throws Exception {
+ final String TOPIC = "test/resiliency";
+ final String SUBSCRIBER_CLIENT_ID = "subscriber";
+ final String PUBLISHER_CLIENT_ID = "publisher";
+ final CountDownLatch pubRelLatch = new CountDownLatch(1);
+ final CountDownLatch pubCompLatch = new CountDownLatch(1);
+ AtomicInteger messageCount = new AtomicInteger(0);
+
+ // Block the outgoing PUBREL so the broker has processed PUBREC but the consumer never receives PUBREL
+ MQTTInterceptor pubRelInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBREL) {
+ pubRelLatch.countDown();
+ logger.info("Blocking outgoing {}", packet.fixedHeader().messageType());
+ return false;
+ }
+ logger.info("Allowing outgoing {}", packet.fixedHeader().messageType());
+ return true;
+ };
+ server.getRemotingService().addOutgoingInterceptor(pubRelInterceptor);
+
+ // Consumer with persistent session
+ MqttClient subscriber = createPahoClient(SUBSCRIBER_CLIENT_ID);
+ subscriber.setCallback(new DefaultMqttCallback() {
+ @Override
+ public void messageArrived(String topic, MqttMessage message) throws Exception {
+ messageCount.incrementAndGet();
+ logger.info("messageArrived({}, {})", topic, message);
+ }
+ });
+ MqttConnectionOptions subscriberOptions = new MqttConnectionOptionsBuilder()
+ .cleanStart(false)
+ .sessionExpiryInterval(300L)
+ .build();
+ subscriber.connect(subscriberOptions);
+ subscriber.subscribe(TOPIC, EXACTLY_ONCE);
+
+ // Producer
+ MqttClient producer = createPahoClient(PUBLISHER_CLIENT_ID);
+ producer.connect();
+
+ producer.publish(TOPIC, RandomUtil.randomBytes(), EXACTLY_ONCE, false);
+
+ Wait.assertEquals(0, () -> getPubCacheSize(PUBLISHER_CLIENT_ID));
+
+ producer.disconnect();
+ producer.close();
+
+ assertNull(getPubCache(PUBLISHER_CLIENT_ID));
+
+ assertTrue(pubRelLatch.await(5, TimeUnit.SECONDS));
+
+ if (restart) {
+ server.stop();
+ waitForServerToStop(server);
+ server.start();
+ waitForServerToStart(server);
+ } else {
+ server.getRemotingService().clearInterceptors();
+ server.getActiveMQServerControl().closeConnectionWithID(server.getActiveMQServerControl().listConnectionIDs()[0]);
+ }
+
+ assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(SUBSCRIBER_CLIENT_ID));
+ assertTrue(getPubRecCache(SUBSCRIBER_CLIENT_ID).contains(ByteUtil.intToBytes(1)));
+
+ MQTTInterceptor pubCompInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBCOMP) {
+ pubCompLatch.countDown();
+ }
+ return true;
+ };
+ server.getRemotingService().addIncomingInterceptor(pubCompInterceptor);
+
+ reconnectSafely(subscriber);
+
+ assertTrue(pubCompLatch.await(5, TimeUnit.SECONDS));
+
+ Wait.assertEquals(0L, () -> getSubscriptionQueue(TOPIC, SUBSCRIBER_CLIENT_ID).getMessageCount(), 500, 25);
+
+ assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(SUBSCRIBER_CLIENT_ID));
+ Wait.assertEquals(0, () -> getPubRecCacheSize(SUBSCRIBER_CLIENT_ID));
+ assertEquals(1, messageCount.get());
+
+ subscriber.disconnect();
+
+ // connect again to clean the session which will completely remove the cache from memory and disk
+ subscriber.connect(new MqttConnectionOptionsBuilder().cleanStart(true).sessionExpiryInterval(0L).build());
+ assertNull(getPubRecCache(SUBSCRIBER_CLIENT_ID));
+ subscriber.disconnect();
+ subscriber.close();
+ }
+
+ /**
+ * Verifies that the broker correctly completes the QoS 2 flow if it sends the PUBREL but fails to receive the
+ * corresponding PUBCOMP before the broker restarts.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2BrokerRestartAfterPubRelSent() throws Exception {
+ testQoS2FailureAfterPubRelSent(true);
+ }
+
+ /**
+ * Same test as {@link testQoS2BrokerRestartAfterPubRelSent} but disconnecting the client instead of restarting the
+ * broker.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2ClientDisconnectAfterPubRelSent() throws Exception {
+ testQoS2FailureAfterPubRelSent(false);
+ }
+
+ public void testQoS2FailureAfterPubRelSent(boolean restart) throws Exception {
+ final String TOPIC = "test/resiliency";
+ final String SUBSCRIBER_CLIENT_ID = "subscriber";
+ final String PUBLISHER_CLIENT_ID = "publisher";
+ final CountDownLatch pubCompBlockedLatch = new CountDownLatch(1);
+ final CountDownLatch stopLatch = new CountDownLatch(1);
+ final CountDownLatch pubCompLatch = new CountDownLatch(1);
+ AtomicInteger messageCount = new AtomicInteger(0);
+
+ // Block the incoming PUBCOMP so the broker has sent PUBREL but never processes the PUBCOMP
+ MQTTInterceptor pubCompBlocker = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBCOMP) {
+ pubCompBlockedLatch.countDown();
+ try {
+ stopLatch.await(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ return false;
+ }
+ return true;
+ };
+ server.getRemotingService().addIncomingInterceptor(pubCompBlocker);
+
+ // Consumer with persistent session
+ MqttClient subscriber = createPahoClient(SUBSCRIBER_CLIENT_ID);
+ subscriber.setCallback(new DefaultMqttCallback() {
+ @Override
+ public void messageArrived(String topic, MqttMessage message) throws Exception {
+ messageCount.incrementAndGet();
+ logger.info("messageArrived({}, {})", topic, message);
+ }
+ });
+ MqttConnectionOptions subscriberOptions = new MqttConnectionOptionsBuilder()
+ .cleanStart(false)
+ .sessionExpiryInterval(300L)
+ .build();
+ subscriber.connect(subscriberOptions);
+ subscriber.subscribe(TOPIC, EXACTLY_ONCE);
+
+ // Producer
+ MqttClient producer = createPahoClient(PUBLISHER_CLIENT_ID);
+ producer.connect();
+
+ producer.publish(TOPIC, RandomUtil.randomBytes(), EXACTLY_ONCE, false);
+
+ producer.disconnect();
+ producer.close();
+
+ assertTrue(pubCompBlockedLatch.await(5, TimeUnit.SECONDS));
+ stopLatch.countDown();
+
+ if (restart) {
+ server.stop();
+ waitForServerToStop(server);
+ server.start();
+ waitForServerToStart(server);
+ } else {
+ server.getRemotingService().clearInterceptors();
+ Wait.assertEquals(0, () -> server.getRemotingService().getConnections().size(), 1000, 10);
+ }
+
+ assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(SUBSCRIBER_CLIENT_ID));
+ assertTrue(getPubRecCache(SUBSCRIBER_CLIENT_ID).contains(ByteUtil.intToBytes(1)));
+
+ MQTTInterceptor pubCompInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBCOMP) {
+ pubCompLatch.countDown();
+ }
+ return true;
+ };
+ server.getRemotingService().addIncomingInterceptor(pubCompInterceptor);
+
+ reconnectSafely(subscriber);
+
+ assertTrue(pubCompLatch.await(5, TimeUnit.SECONDS));
+
+ Wait.assertEquals(0L, () -> getSubscriptionQueue(TOPIC, SUBSCRIBER_CLIENT_ID).getMessageCount(), 500, 25);
+
+ assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(SUBSCRIBER_CLIENT_ID));
+ Wait.assertEquals(0, () -> getPubRecCacheSize(SUBSCRIBER_CLIENT_ID));
+ assertEquals(1, messageCount.get());
+
+ subscriber.disconnect();
+
+ // connect again to clean the session which will completely remove the cache from memory and disk
+ subscriber.connect(new MqttConnectionOptionsBuilder().cleanStart(true).sessionExpiryInterval(0L).build());
+ assertNull(getPubRecCache(SUBSCRIBER_CLIENT_ID));
+ subscriber.disconnect();
+ subscriber.close();
+ }
+
+ /**
+ * Verifies that after a complete QoS 2 protocol exchange (PUBLISH, PUBREC, PUBREL, PUBCOMP), a broker restart does
+ * not cause re-delivery.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2BrokerRestartAfterPubCompSent() throws Exception {
+ testQoS2FailureAfterPubCompSent(true);
+ }
+
+ /**
+ * Same test as {@link testQoS2BrokerRestartAfterPubCompSent} but disconnecting the client instead of restarting the
+ * broker.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2ClientDisconnectAfterPubCompSent() throws Exception {
+ testQoS2FailureAfterPubCompSent(false);
+ }
+
+ public void testQoS2FailureAfterPubCompSent(boolean restart) throws Exception {
+ final String TOPIC = "test/resiliency";
+ final String SUBSCRIBER_CLIENT_ID = "subscriber";
+ final String PUBLISHER_CLIENT_ID = "publisher";
+ final CountDownLatch pubCompLatch = new CountDownLatch(1);
+ AtomicInteger messageCount = new AtomicInteger(0);
+
+ // Consumer with persistent session
+ MqttClient subscriber = createPahoClient(SUBSCRIBER_CLIENT_ID);
+ subscriber.setCallback(new DefaultMqttCallback() {
+ @Override
+ public void messageArrived(String topic, MqttMessage message) throws Exception {
+ messageCount.incrementAndGet();
+ logger.info("messageArrived({}, {})", topic, message);
+ }
+ });
+ MqttConnectionOptions subscriberOptions = new MqttConnectionOptionsBuilder()
+ .cleanStart(false)
+ .sessionExpiryInterval(300L)
+ .build();
+ subscriber.connect(subscriberOptions);
+ subscriber.subscribe(TOPIC, EXACTLY_ONCE);
+
+ // Track PUBCOMPs to know when both QoS 2 flows are complete
+ MQTTInterceptor pubCompInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBCOMP) {
+ pubCompLatch.countDown();
+ }
+ return true;
+ };
+ server.getRemotingService().addIncomingInterceptor(pubCompInterceptor);
+
+ // Producer
+ MqttClient producer = createPahoClient(PUBLISHER_CLIENT_ID);
+ producer.connect();
+
+ producer.publish(TOPIC, RandomUtil.randomBytes(), EXACTLY_ONCE, false);
+
+ producer.disconnect();
+ producer.close();
+
+ // Wait for both QoS 2 flows to complete
+ assertTrue(pubCompLatch.await(5, TimeUnit.SECONDS));
+ Wait.assertEquals(0L, () -> getSubscriptionQueue(TOPIC, SUBSCRIBER_CLIENT_ID).getMessageCount(), 500, 25);
+
+ assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(SUBSCRIBER_CLIENT_ID));
+ assertEquals(0, getPubRecCacheSize(SUBSCRIBER_CLIENT_ID));
+
+ if (restart) {
+ server.stop();
+ waitForServerToStop(server);
+ server.start();
+ waitForServerToStart(server);
+ } else {
+ server.getRemotingService().clearInterceptors();
+ server.getActiveMQServerControl().closeConnectionWithID(
+ server.getActiveMQServerControl().listConnectionIDs()[0]);
+ }
+
+ int countBeforeReconnect = messageCount.get();
+ reconnectSafely(subscriber);
+
+ // Verify no unexpected re-delivery
+ assertFalse(Wait.waitFor(() -> messageCount.get() > countBeforeReconnect, 500, 25), "Unexpected message delivered after restart");
+ Wait.assertEquals(0L, () -> getSubscriptionQueue(TOPIC, SUBSCRIBER_CLIENT_ID).getMessageCount(), 500, 25);
+
+ assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(SUBSCRIBER_CLIENT_ID));
+ if (restart) {
+ assertNull(getPubRecCache(SUBSCRIBER_CLIENT_ID));
+ } else {
+ assertEquals(0, getPubRecCacheSize(SUBSCRIBER_CLIENT_ID));
+ }
+
+ subscriber.disconnect();
+
+ // connect again to clean the session which will completely remove the cache from memory and disk
+ subscriber.connect(new MqttConnectionOptionsBuilder().cleanStart(true).sessionExpiryInterval(0L).build());
+ assertNull(getPubRecCache(SUBSCRIBER_CLIENT_ID));
+ subscriber.disconnect();
+ subscriber.close();
+ }
+
+ /**
+ * Verifies that incomplete QoS 2 flow state (packet ID correlations) is removed when a client with session expiry
+ * interval 0 is disconnected before PUBREC is processed.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2PacketIdCorrelationsRemovedOnSessionExpiryZeroDisconnect() throws Exception {
+ final String TOPIC = "test/resiliency";
+ final String SUBSCRIBER_CLIENT_ID = "subscriber";
+ final String PUBLISHER_CLIENT_ID = "publisher";
+
+ // Block the incoming PUBREC so the packet ID correlation is never removed by the normal QoS 2 flow.
+ // Returning false from an incoming interceptor disconnects the client.
+ final CountDownLatch pubRecLatch = new CountDownLatch(1);
+ MQTTInterceptor pubRecInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBREC) {
+ pubRecLatch.countDown();
+ return false;
+ }
+ return true;
+ };
+ server.getRemotingService().addIncomingInterceptor(pubRecInterceptor);
+
+ MqttClient subscriber = createPahoClient(SUBSCRIBER_CLIENT_ID);
+ subscriber.setCallback(new DefaultMqttCallback() {
+ @Override
+ public void messageArrived(String topic, MqttMessage message) throws Exception {
+ }
+ });
+ subscriber.connect(new MqttConnectionOptionsBuilder().cleanStart(true).sessionExpiryInterval(0L).build());
+ subscriber.subscribe(TOPIC, EXACTLY_ONCE);
+
+ MqttClient producer = createPahoClient(PUBLISHER_CLIENT_ID);
+ producer.connect();
+ producer.publish(TOPIC, RandomUtil.randomBytes(), EXACTLY_ONCE, false);
+ Wait.assertEquals(0, () -> getPubCacheSize(PUBLISHER_CLIENT_ID));
+ producer.disconnect();
+ producer.close();
+
+ // Wait for PUBREC to be intercepted (which disconnects the subscriber)
+ assertTrue(pubRecLatch.await(5, TimeUnit.SECONDS));
+ server.getRemotingService().clearInterceptors();
+ Wait.assertEquals(0, () -> server.getRemotingService().getConnections().size(), 1000, 10);
+
+ // Since session expiry is 0, the session was cleaned up on disconnect.
+ // The packet ID correlations should have been removed.
+ assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(SUBSCRIBER_CLIENT_ID));
+
+ subscriber.close();
+ }
+
+ /**
+ * Verifies that incomplete QoS 2 flow state (packet ID correlations) is removed when a session expires via the
+ * session expiry scanner.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2PacketIdCorrelationsRemovedOnSessionExpiry() throws Exception {
+ final String TOPIC = "test/resiliency";
+ final String SUBSCRIBER_CLIENT_ID = "subscriber";
+ final String PUBLISHER_CLIENT_ID = "publisher";
+
+ final CountDownLatch pubRecLatch = new CountDownLatch(1);
+ MQTTInterceptor pubRecInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBREC) {
+ pubRecLatch.countDown();
+ return false;
+ }
+ return true;
+ };
+ server.getRemotingService().addIncomingInterceptor(pubRecInterceptor);
+
+ MqttClient subscriber = createPahoClient(SUBSCRIBER_CLIENT_ID);
+ subscriber.setCallback(new DefaultMqttCallback() {
+ @Override
+ public void messageArrived(String topic, MqttMessage message) throws Exception {
+ }
+ });
+ subscriber.connect(new MqttConnectionOptionsBuilder().cleanStart(true).sessionExpiryInterval(1L).build());
+ subscriber.subscribe(TOPIC, EXACTLY_ONCE);
+
+ MqttClient producer = createPahoClient(PUBLISHER_CLIENT_ID);
+ producer.connect();
+ producer.publish(TOPIC, RandomUtil.randomBytes(), EXACTLY_ONCE, false);
+ Wait.assertEquals(0, () -> getPubCacheSize(PUBLISHER_CLIENT_ID));
+ producer.disconnect();
+ producer.close();
+
+ assertTrue(pubRecLatch.await(5, TimeUnit.SECONDS));
+ server.getRemotingService().clearInterceptors();
+ Wait.assertEquals(0, () -> server.getRemotingService().getConnections().size(), 1000, 10);
+
+ // Packet ID correlations should exist since PUBREC was never processed
+ assertTrue(getProtocolManager().getStateManager().getPacketIdCorrelationSize(SUBSCRIBER_CLIENT_ID) > 0);
+
+ // Wait for the session to expire then trigger the scanner
+ Thread.sleep(1500);
+ scanSessions();
+
+ // The session expiry scanner should have cleaned up the packet ID correlations
+ assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(SUBSCRIBER_CLIENT_ID));
+ assertNull(getSessionStates().get(SUBSCRIBER_CLIENT_ID));
+
+ subscriber.close();
+ }
+
+ /**
+ * Verifies that incomplete QoS 2 flow state (packet ID correlations) is removed when a client reconnects with
+ * clean start = true.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2PacketIdCorrelationsRemovedOnCleanStartReconnect() throws Exception {
+ final String TOPIC = "test/resiliency";
+ final String SUBSCRIBER_CLIENT_ID = "subscriber";
+ final String PUBLISHER_CLIENT_ID = "publisher";
+
+ final CountDownLatch pubRecLatch = new CountDownLatch(1);
+ MQTTInterceptor pubRecInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBREC) {
+ pubRecLatch.countDown();
+ return false;
+ }
+ return true;
+ };
+ server.getRemotingService().addIncomingInterceptor(pubRecInterceptor);
+
+ MqttClient subscriber = createPahoClient(SUBSCRIBER_CLIENT_ID);
+ subscriber.setCallback(new DefaultMqttCallback() {
+ @Override
+ public void messageArrived(String topic, MqttMessage message) throws Exception {
+ }
+ });
+ subscriber.connect(new MqttConnectionOptionsBuilder().cleanStart(true).sessionExpiryInterval(300L).build());
+ subscriber.subscribe(TOPIC, EXACTLY_ONCE);
+
+ MqttClient producer = createPahoClient(PUBLISHER_CLIENT_ID);
+ producer.connect();
+ producer.publish(TOPIC, RandomUtil.randomBytes(), EXACTLY_ONCE, false);
+ Wait.assertEquals(0, () -> getPubCacheSize(PUBLISHER_CLIENT_ID));
+ producer.disconnect();
+ producer.close();
+
+ assertTrue(pubRecLatch.await(5, TimeUnit.SECONDS));
+ server.getRemotingService().clearInterceptors();
+ Wait.assertEquals(0, () -> server.getRemotingService().getConnections().size(), 1000, 10);
+
+ // Packet ID correlations should exist since PUBREC was never processed
+ assertTrue(getProtocolManager().getStateManager().getPacketIdCorrelationSize(SUBSCRIBER_CLIENT_ID) > 0);
+
+ // Reconnect with clean start to discard previous session state
+ subscriber.connect(new MqttConnectionOptionsBuilder().cleanStart(true).sessionExpiryInterval(0L).build());
+
+ // The clean start should have cleared the packet ID correlations
+ assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(SUBSCRIBER_CLIENT_ID));
+
+ subscriber.disconnect();
+ subscriber.close();
+ }
+}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/spec/QoSTests.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/spec/QoSTests.java
index 92ff22721d..80fe0da0ae 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/spec/QoSTests.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/spec/QoSTests.java
@@ -18,6 +18,7 @@ package org.apache.activemq.artemis.tests.integration.mqtt5.spec;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
@@ -27,6 +28,19 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
+import com.hivemq.client.mqtt.datatypes.MqttQos;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5Client;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5ClientConfig;
+import com.hivemq.client.mqtt.mqtt5.advanced.interceptor.qos2.Mqtt5IncomingQos2Interceptor;
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
+import com.hivemq.client.mqtt.mqtt5.message.publish.pubcomp.Mqtt5PubCompBuilder;
+import com.hivemq.client.mqtt.mqtt5.message.publish.pubrec.Mqtt5PubRecBuilder;
+import com.hivemq.client.mqtt.mqtt5.message.publish.pubrec.Mqtt5PubRecReasonCode;
+import com.hivemq.client.mqtt.mqtt5.message.publish.pubrel.Mqtt5PubRel;
+import io.reactivex.schedulers.Schedulers;
+import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPubAckMessage;
import io.netty.handler.codec.mqtt.MqttPubReplyMessageVariableHeader;
@@ -38,6 +52,7 @@ import org.apache.activemq.artemis.core.protocol.mqtt.MQTTReasonCodes;
import org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.tests.integration.mqtt5.MQTT5TestSupport;
+import org.apache.activemq.artemis.utils.ByteUtil;
import org.apache.activemq.artemis.utils.RandomUtil;
import org.apache.activemq.artemis.tests.util.Wait;
import org.eclipse.paho.mqttv5.client.MqttClient;
@@ -368,13 +383,8 @@ public class QoSTests extends MQTT5TestSupport {
MQTTInterceptor incomingInterceptor = (packet, connection) -> {
if (packet.fixedHeader().messageType() == MqttMessageType.PUBCOMP) {
- try {
- // ensure the message is still in the management queue before we get the PUBCOMP from the client
- Wait.assertEquals(1L, () -> server.locateQueue(MQTTUtil.QOS2_MANAGEMENT_QUEUE_PREFIX + CONSUMER_ID).getMessageCount(), 2000, 100);
- Wait.assertEquals(1L, () -> server.locateQueue(MQTTUtil.QOS2_MANAGEMENT_QUEUE_PREFIX + CONSUMER_ID).getDeliveringCount(), 2000, 100);
- } catch (Exception e) {
- return false;
- }
+ // ensure the packet ID is stored in the PUBREC cache before we get the PUBCOMP from the client
+ assertTrue(getPubRecCache(CONSUMER_ID).contains(ByteUtil.intToBytes(packetId.get())));
// ensure the ids match so we know this is the "corresponding" PUBCOMP for the previous PUBLISH
assertEquals(packetId.get(), ((MqttPubReplyMessageVariableHeader)packet.variableHeader()).messageId());
@@ -691,4 +701,96 @@ public class QoSTests extends MQTT5TestSupport {
assertTrue(ackLatch.await(messageExpiryInterval * 2, TimeUnit.SECONDS));
Wait.assertEquals(1, () -> getSubscriptionQueue(TOPIC, CONSUMER_ID).getMessagesExpired());
}
+
+ /*
+ * [MQTT-4.3.3-4] In the QoS 2 delivery protocol, the sender MUST send a PUBREL packet when it receives a PUBREC
+ * packet from the receiver with a Reason Code value less than 0x80.
+ *
+ * The converse: when the broker (sender) receives a PUBREC with Reason Code >= 0x80, it MUST NOT send PUBREL.
+ * Instead, it should treat the PUBLISH as acknowledged (i.e. discard it without sending PUBREL).
+ *
+ * This test uses the HiveMQ MQTT client as the Paho client doesn't support the ability to simulate a PUBREC with
+ * a reason code greater than 0x80.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testQoS2PubRecError() throws Exception {
+ final String TOPIC = RandomUtil.randomUUIDString();
+ final String CONSUMER_ID = "consumer";
+ final AtomicBoolean pubRelSent = new AtomicBoolean(false);
+ final CountDownLatch pubRecReceived = new CountDownLatch(1);
+ final AtomicBoolean correlationExistedDuringDelivery = new AtomicBoolean(false);
+
+ MQTTInterceptor incomingInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBREC) {
+ pubRecReceived.countDown();
+ }
+ return true;
+ };
+
+ MQTTInterceptor outgoingInterceptor = (packet, connection) -> {
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBLISH) {
+ correlationExistedDuringDelivery.set(getProtocolManager().getStateManager().getPacketIdCorrelationSize(CONSUMER_ID) > 0);
+ }
+ if (packet.fixedHeader().messageType() == MqttMessageType.PUBREL) {
+ pubRelSent.set(true);
+ }
+ return true;
+ };
+
+ server.getRemotingService().addIncomingInterceptor(incomingInterceptor);
+ server.getRemotingService().addOutgoingInterceptor(outgoingInterceptor);
+
+ EpollEventLoopGroup hivemqEventLoop = new EpollEventLoopGroup();
+ try {
+ Mqtt5BlockingClient consumer = Mqtt5Client.builder()
+ .identifier(CONSUMER_ID)
+ .serverHost("localhost")
+ .serverPort(getPort())
+ .executorConfig()
+ .nettyExecutor(hivemqEventLoop)
+ .applyExecutorConfig()
+ .advancedConfig()
+ .interceptors()
+ .incomingQos2Interceptor(new Mqtt5IncomingQos2Interceptor() {
+ @Override
+ public void onPublish(Mqtt5ClientConfig clientConfig, Mqtt5Publish publish, Mqtt5PubRecBuilder pubRecBuilder) {
+ pubRecBuilder.reasonCode(Mqtt5PubRecReasonCode.UNSPECIFIED_ERROR);
+ }
+
+ @Override
+ public void onPubRel(Mqtt5ClientConfig clientConfig, Mqtt5PubRel pubRel, Mqtt5PubCompBuilder pubCompBuilder) {
+ }
+ })
+ .applyInterceptors()
+ .applyAdvancedConfig()
+ .buildBlocking();
+ consumer.connect();
+ Mqtt5BlockingClient.Mqtt5Publishes publishes = consumer.publishes(MqttGlobalPublishFilter.ALL);
+ consumer.subscribeWith()
+ .topicFilter(TOPIC)
+ .qos(MqttQos.EXACTLY_ONCE)
+ .send();
+
+ MqttClient producer = createPahoClient("producer");
+ producer.connect();
+ producer.publish(TOPIC, RandomUtil.randomUUIDString().getBytes(), 2, false);
+ producer.disconnect();
+ producer.close();
+
+ assertTrue(pubRecReceived.await(5, TimeUnit.SECONDS), "PUBREC was not received by the broker");
+ assertTrue(correlationExistedDuringDelivery.get(), "Packet ID correlation should exist during delivery");
+ Wait.assertEquals(0L, () -> getSubscriptionQueue(TOPIC, CONSUMER_ID).getMessageCount(), 2000, 100);
+ Wait.assertEquals(1L, () -> getSubscriptionQueue(TOPIC, CONSUMER_ID).getMessagesAcknowledged(), 2000, 100);
+ Wait.assertEquals(0, () -> getProtocolManager().getStateManager().getPacketIdCorrelationSize(CONSUMER_ID), 2000, 100);
+ assertFalse(pubRelSent.get(), "PUBREL should not be sent in response to an error PUBREC");
+ assertNull(getPubRecCache(CONSUMER_ID));
+
+ publishes.close();
+ consumer.disconnect();
+ } finally {
+ hivemqEventLoop.shutdownGracefully(0, 0, TimeUnit.MILLISECONDS).sync();
+ Schedulers.shutdown();
+ }
+ }
}
diff --git a/tests/soak-tests/pom.xml b/tests/soak-tests/pom.xml
index 5f5ea7aa2f..078ddbfeb7 100644
--- a/tests/soak-tests/pom.xml
+++ b/tests/soak-tests/pom.xml
@@ -188,7 +188,11 @@
<artifactId>spring-jms</artifactId>
<scope>test</scope>
</dependency>
-
+ <dependency>
+ <groupId>com.hivemq</groupId>
+ <artifactId>hivemq-mqtt-client</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
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
new file mode 100644
index 0000000000..99a7c753ce
--- /dev/null
+++ b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/mqtt/resiliency/QoS2CombinedResiliencySoakTest.java
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.soak.mqtt.resiliency;
+
+import java.lang.invoke.MethodHandles;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
+import com.hivemq.client.mqtt.datatypes.MqttQos;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
+import org.apache.activemq.artemis.utils.TestParameters;
+import org.apache.activemq.artemis.utils.Wait;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class QoS2CombinedResiliencySoakTest extends QoS2ResiliencySoakTestSupport {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static final String TEST_NAME = "QOS2_COMBINED_RESILIENCY_SOAK";
+
+ private static final int NUM_PUBLISHERS = TestParameters.testProperty(TEST_NAME, "NUM_PUBLISHERS", 5);
+ private static final int NUM_SUBSCRIBERS = TestParameters.testProperty(TEST_NAME, "NUM_SUBSCRIBERS", 5);
+ private static final int NUM_MESSAGES = TestParameters.testProperty(TEST_NAME, "NUM_MESSAGES", 2_000);
+ private static final int RESTART_PAUSE = TestParameters.testProperty(TEST_NAME, "RESTART_PAUSE", 2_000);
+ private static final int TIMEOUT_SECONDS = TestParameters.testProperty(TEST_NAME, "TIMEOUT_SECONDS", 180);
+
+ @Test
+ @Timeout(value = 5, unit = TimeUnit.MINUTES)
+ public void testQoS2CombinedResiliency() throws Exception {
+ disableProtocolLogging();
+ final String PUB_CLIENT_ID_PREFIX = "pub-";
+ final String SUB_CLIENT_ID_PREFIX = "sub-";
+
+ logger.info("{} publishers, {} subscribers, {} messages/publisher, {} total expected per subscriber", NUM_PUBLISHERS, NUM_SUBSCRIBERS, NUM_MESSAGES, NUM_PUBLISHERS * NUM_MESSAGES);
+
+ // create and subscribe all subscribers
+ final List<Mqtt5BlockingClient> subscribers = new ArrayList<>(NUM_SUBSCRIBERS);
+ runAfter(() -> subscribers.forEach(c -> {
+ try {
+ c.disconnect();
+ } catch (Exception ignored) {
+ }
+ }));
+
+ final Map<String, Set<String>> receivedPerSubscriber = new HashMap<>();
+ final Map<String, Set<String>> duplicatesPerSubscriber = new HashMap<>();
+ final AtomicLong lastReceiveTime = new AtomicLong(System.currentTimeMillis());
+
+ for (int i = 0; i < NUM_SUBSCRIBERS; i++) {
+ String clientId = SUB_CLIENT_ID_PREFIX + i;
+ Mqtt5BlockingClient subscriber = createHiveMQClient(clientId, true);
+ final Set<String> received = ConcurrentHashMap.newKeySet(NUM_PUBLISHERS * NUM_MESSAGES);
+ receivedPerSubscriber.put(clientId, received);
+ final Set<String> duplicates = ConcurrentHashMap.newKeySet();
+ duplicatesPerSubscriber.put(clientId, duplicates);
+ subscriber.toAsync().publishes(MqttGlobalPublishFilter.ALL, publish -> {
+ String payload = new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8);
+ if (!received.add(payload)) {
+ logger.warn("Subscriber {} received duplicate: {}", clientId, payload);
+ duplicates.add(payload);
+ }
+ lastReceiveTime.set(System.currentTimeMillis());
+ });
+ subscriber.connectWith()
+ .cleanStart(false)
+ .sessionExpiryInterval(300)
+ .send();
+ subscriber.subscribeWith()
+ .topicFilter(TOPIC)
+ .qos(MqttQos.EXACTLY_ONCE)
+ .send();
+ subscribers.add(subscriber);
+ assertNotNull(getSubscriptionQueue(TOPIC, clientId));
+ logger.info("Subscriber {} connected and subscribed", clientId);
+ }
+
+ // create and connect publishers
+ final List<Mqtt5BlockingClient> publishers = new ArrayList<>();
+ runAfter(() -> publishers.forEach(c -> {
+ try {
+ c.disconnect();
+ } catch (Exception ignored) {
+ }
+ }));
+ for (int i = 0; i < NUM_PUBLISHERS; i++) {
+ String clientId = PUB_CLIENT_ID_PREFIX + i;
+ Mqtt5BlockingClient publisher = createHiveMQClient(clientId, true);
+ publisher.connectWith()
+ .cleanStart(false)
+ .sessionExpiryInterval(300)
+ .send();
+ publishers.add(publisher);
+ logger.info("Publisher {} connected", clientId);
+ }
+
+ // start publisher tasks
+ PublishResult publishResult = startPublishing(publishers, NUM_MESSAGES);
+
+ // start broker restart task
+ BrokerRestartTask restartTask = startBrokerRestartTask(RESTART_PAUSE, () -> {
+ logger.info("===========");
+ for (Map.Entry<String, Set<String>> entry : receivedPerSubscriber.entrySet()) {
+ logger.info("Subscriber {} received {}/{} messages", entry.getKey(), entry.getValue().size(), NUM_PUBLISHERS * NUM_MESSAGES);
+ }
+ logger.info("Last message received {}ms ago.", System.currentTimeMillis() - lastReceiveTime.get());
+ logger.info("===========");
+ });
+
+ // wait for all publishers to finish sending
+ publishResult.executor().shutdown();
+ assertTrue(publishResult.executor().awaitTermination(TIMEOUT_SECONDS, TimeUnit.SECONDS), "Publishers did not finish in time");
+ logger.info("All publishers finished. Total messages sent: {}. Publish errors: {}", publishResult.sentMessages().size(), publishResult.publishErrors().get());
+
+ // wait for all subscribers to receive all messages
+ final long STALL_TIMEOUT_MS = 20_000;
+ Wait.assertTrue(() -> {
+ if (System.currentTimeMillis() - lastReceiveTime.get() > STALL_TIMEOUT_MS) {
+ for (Map.Entry<String, Set<String>> entry : receivedPerSubscriber.entrySet()) {
+ logger.warn("Subscriber {} received {}/{} messages", entry.getKey(), entry.getValue().size(), NUM_PUBLISHERS * NUM_MESSAGES);
+ }
+ throw new AssertionError("No subscriber has received a message in " + STALL_TIMEOUT_MS / 1000 + " seconds");
+ }
+ for (Map.Entry<String, Set<String>> duplicates : duplicatesPerSubscriber.entrySet()) {
+ assertEquals(0, duplicates.getValue().size(), "Subscriber " + duplicates.getKey() + " received duplicates: " + duplicates.getValue());
+ }
+ for (Set<String> received : receivedPerSubscriber.values()) {
+ if (received.size() < NUM_PUBLISHERS * NUM_MESSAGES) {
+ return false;
+ }
+ }
+ return true;
+ }, TIMEOUT_SECONDS * 1000L, 100);
+
+ disableProtocolLogging();
+
+ // stop reconnection task, ensure broker is running
+ stopBrokerRestartTask(restartTask);
+
+ // verify all expected messages received with no duplicates
+ for (Mqtt5BlockingClient subscriber : subscribers) {
+ 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");
+ assertEquals(0, getProtocolManager().getStateManager().getPacketIdCorrelationSize(clientId));
+ assertEquals(0, getSubCacheSize(clientId));
+ cleanDisconnect(subscriber);
+ assertNull(getSubCache(clientId), "Sub cache should be null after clean start for " + clientId);
+ }
+
+ for (Mqtt5BlockingClient publisher : publishers) {
+ String clientId = getClientId(publisher);
+ Wait.assertEquals(0, () -> getPubCacheSize(clientId), 5000, 100);
+ cleanDisconnect(publisher);
+ assertNull(getPubCache(clientId), "Pub cache should be null after clean start for " + clientId);
+ }
+ }
+}
\ No newline at end of file
diff --git a/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/mqtt/resiliency/QoS2PublisherResiliencySoakTest.java b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/mqtt/resiliency/QoS2PublisherResiliencySoakTest.java
new file mode 100644
index 0000000000..ac6dd3e642
--- /dev/null
+++ b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/mqtt/resiliency/QoS2PublisherResiliencySoakTest.java
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.soak.mqtt.resiliency;
+
+import java.lang.invoke.MethodHandles;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
+import com.hivemq.client.mqtt.datatypes.MqttQos;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
+import org.apache.activemq.artemis.utils.TestParameters;
+import org.apache.activemq.artemis.utils.Wait;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class QoS2PublisherResiliencySoakTest extends QoS2ResiliencySoakTestSupport {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static final String TEST_NAME = "QOS2_PUBLISHER_RESILIENCY_SOAK";
+
+ private static final int NUM_PUBLISHERS = TestParameters.testProperty(TEST_NAME, "NUM_PUBLISHERS", 5);
+ private static final int NUM_MESSAGES = TestParameters.testProperty(TEST_NAME, "NUM_MESSAGES", 10_000);
+ private static final int RESTART_PAUSE = TestParameters.testProperty(TEST_NAME, "RESTART_PAUSE", 2_000);
+ private static final int TIMEOUT_SECONDS = TestParameters.testProperty(TEST_NAME, "TIMEOUT_SECONDS", 180);
+
+ @Test
+ @Timeout(value = 5, unit = TimeUnit.MINUTES)
+ public void testQoS2PublisherResiliency() throws Exception {
+ disableProtocolLogging();
+ final String PUB_CLIENT_ID_PREFIX = "pub-";
+ final String SUB_CLIENT_ID = "sub";
+
+ // create subscription queue for consuming messages later
+ Mqtt5BlockingClient subscriber = createHiveMQClient(SUB_CLIENT_ID, false);
+ subscriber.connectWith()
+ .cleanStart(false)
+ .sessionExpiryInterval(300)
+ .send();
+ subscriber.subscribeWith()
+ .topicFilter(TOPIC)
+ .qos(MqttQos.AT_LEAST_ONCE)
+ .send();
+ subscriber.disconnect();
+
+ assertNotNull(getSubscriptionQueue(TOPIC, SUB_CLIENT_ID));
+
+ logger.info("{} publishers, {} messages/publisher, {} total expected", NUM_PUBLISHERS, NUM_MESSAGES, NUM_PUBLISHERS * NUM_MESSAGES);
+
+ // create and connect publishers
+ final List<Mqtt5BlockingClient> publishers = new ArrayList<>();
+ runAfter(() -> publishers.forEach(c -> {
+ try {
+ c.disconnect();
+ } catch (Exception ignored) {
+ }
+ }));
+ for (int i = 0; i < NUM_PUBLISHERS; i++) {
+ String clientId = PUB_CLIENT_ID_PREFIX + i;
+ Mqtt5BlockingClient publisher = createHiveMQClient(clientId, true);
+ publisher.connectWith()
+ .cleanStart(false)
+ .sessionExpiryInterval(300)
+ .send();
+ publishers.add(publisher);
+ logger.info("Publisher {} connected", clientId);
+ }
+
+ // start broker restart task
+ BrokerRestartTask restartTask = startBrokerRestartTask(RESTART_PAUSE, () -> {
+ logger.info("===========");
+ logger.info("Subscription queue received {}/{} messages", getSubscriptionQueue(TOPIC, SUB_CLIENT_ID).getMessageCount(), NUM_PUBLISHERS * NUM_MESSAGES);
+ logger.info("===========");
+ });
+
+ // enableProtocolLogging();
+
+ // start publisher tasks
+ PublishResult publishResult = startPublishing(publishers, NUM_MESSAGES);
+
+ // wait for all publishers to finish
+ publishResult.executor().shutdown();
+ assertTrue(publishResult.executor().awaitTermination(TIMEOUT_SECONDS, TimeUnit.SECONDS), "Publishers did not finish in time");
+ logger.info("All publishers finished. Total messages sent: {}. Publish errors: {}", publishResult.sentMessages().size(), publishResult.publishErrors().get());
+
+ disableProtocolLogging();
+
+ // stop restart task and ensure broker is running
+ stopBrokerRestartTask(restartTask);
+
+ final long messageCount = getSubscriptionQueue(TOPIC, SUB_CLIENT_ID).getMessageCount();
+
+ // reconnect subscriber to verify there are no duplicates
+ Set<String> consumedMessages = ConcurrentHashMap.newKeySet(NUM_MESSAGES * NUM_PUBLISHERS);
+ AtomicInteger duplicateCount = new AtomicInteger(0);
+ subscriber.toAsync().publishes(MqttGlobalPublishFilter.ALL, publish -> {
+ String payload = new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8);
+ if (consumedMessages.contains(payload)) {
+ logger.warn("Duplicate message: {}", payload);
+ duplicateCount.incrementAndGet();
+ }
+ consumedMessages.add(payload);
+ publishResult.sentMessages().remove(payload);
+ });
+
+ subscriber.connectWith()
+ .cleanStart(false)
+ .sessionExpiryInterval(300)
+ .send();
+
+ Wait.waitFor(() -> consumedMessages.size() == messageCount);
+
+ cleanDisconnect(subscriber);
+
+ assertEquals(0, duplicateCount.get());
+ assertEquals(0, publishResult.sentMessages().size(), "These messages were published, but were not on the broker: " + publishResult.sentMessages());
+ assertEquals(NUM_PUBLISHERS * NUM_MESSAGES, consumedMessages.size());
+
+ for (Mqtt5BlockingClient publisher : publishers) {
+ String clientId = getClientId(publisher);
+ Wait.assertEquals(0, () -> getPubCacheSize(clientId), 5000, 100);
+ cleanDisconnect(publisher);
+ assertNull(getPubCache(clientId), "Pub cache should be null after clean start for " + clientId);
+ }
+ }
+}
\ No newline at end of file
diff --git a/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/mqtt/resiliency/QoS2ResiliencySoakTestSupport.java b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/mqtt/resiliency/QoS2ResiliencySoakTestSupport.java
new file mode 100644
index 0000000000..a3e39805d6
--- /dev/null
+++ b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/mqtt/resiliency/QoS2ResiliencySoakTestSupport.java
@@ -0,0 +1,274 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.soak.mqtt.resiliency;
+
+import java.lang.invoke.MethodHandles;
+import java.nio.charset.StandardCharsets;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.hivemq.client.mqtt.datatypes.MqttQos;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5Client;
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5PublishResult;
+import io.reactivex.schedulers.Schedulers;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds;
+import org.apache.activemq.artemis.core.postoffice.DuplicateIDCache;
+import org.apache.activemq.artemis.core.protocol.mqtt.MQTTProtocolManager;
+import org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil;
+import org.apache.activemq.artemis.core.protocol.mqtt.PacketIdCache;
+import org.apache.activemq.artemis.core.remoting.impl.AbstractAcceptor;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.spi.core.protocol.ProtocolManager;
+import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.activemq.artemis.utils.Wait;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.config.LoggerConfig;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.activemq.artemis.cli.commands.tools.journal.CompactJournal.compactJournal;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalStorageManager.ACTIVEMQ_DATA;
+import static org.apache.activemq.artemis.core.protocol.mqtt.MQTTProtocolManagerFactory.MQTT_PROTOCOL_NAME;
+
+/**
+ * All tests which extend this use the HiveMQ MQTT client because it is the most robust with regard to QoS2 message
+ * flows and error handling. The Paho client had issues which made it unacceptable for these tests.
+ */
+public class QoS2ResiliencySoakTestSupport extends ActiveMQTestBase {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ protected static final String TOPIC = "qos2/resiliency";
+ protected static final int MQTT_PORT = 1883;
+
+ protected ActiveMQServer server;
+
+ @BeforeEach
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ Schedulers.start();
+ server = createServer(true, createDefaultConfig(true));
+ server.getConfiguration().setJournalMinFiles(10).setJournalFileSize(25 * 1024 * 1024);
+ server.getConfiguration().addAcceptorConfiguration(MQTT_PROTOCOL_NAME, "tcp://localhost:" + MQTT_PORT + "?protocols=MQTT");
+ server.getConfiguration().setMqttSessionScanInterval(200);
+
+ server.start();
+ server.waitForActivation(10, TimeUnit.SECONDS);
+ }
+
+ @AfterEach
+ @Override
+ public void tearDown() throws Exception {
+ if (server != null && server.isStarted()) {
+ server.stop();
+ }
+ Schedulers.shutdown();
+ super.tearDown();
+ }
+
+ private static void enableProtocolLogging() {
+ LoggerContext ctx = (LoggerContext) LogManager.getContext(false);
+ org.apache.logging.log4j.core.config.Configuration config = ctx.getConfiguration();
+ LoggerConfig loggerConfig = new LoggerConfig(MQTTUtil.class.getName(), Level.TRACE, true);
+ config.addLogger(MQTTUtil.class.getName(), loggerConfig);
+ ctx.updateLoggers();
+ }
+
+ protected static void disableProtocolLogging() {
+ LoggerContext ctx = (LoggerContext) LogManager.getContext(false);
+ org.apache.logging.log4j.core.config.Configuration config = ctx.getConfiguration();
+ config.removeLogger(MQTTUtil.class.getName());
+ ctx.updateLoggers();
+ }
+
+ protected Set<String> getMissingMessages(Set<String> expected, Set<String> received) {
+ Set<String> missing = new HashSet<>(expected);
+ missing.removeAll(received);
+ return missing;
+ }
+
+ protected static String getClientId(Mqtt5BlockingClient subscriber) {
+ return subscriber.getConfig().getClientIdentifier().get().toString();
+ }
+
+ protected Mqtt5BlockingClient createHiveMQClient(String clientId, boolean autoReconnect) {
+ var builder = Mqtt5Client.builder()
+ .identifier(clientId)
+ .serverHost("localhost")
+ .serverPort(MQTT_PORT);
+ if (autoReconnect) {
+ builder.automaticReconnect()
+ .initialDelay(500, TimeUnit.MILLISECONDS)
+ .maxDelay(500, TimeUnit.MILLISECONDS)
+ .applyAutomaticReconnect();
+ }
+ return builder.buildBlocking();
+ }
+
+ protected static void waitForClientConnected(Mqtt5BlockingClient client) {
+ Wait.waitFor(() -> client.getConfig().getState().isConnected(), 30_000, 500);
+ }
+
+ protected static void cleanDisconnect(Mqtt5BlockingClient client) {
+ logger.info("cleanDisconnect for {}", getClientId(client));
+ try {
+ if (client.getConfig().getState().isConnected()) {
+ client.disconnect();
+ }
+ client.connectWith().cleanStart(true).sessionExpiryInterval(0).send();
+ client.disconnect();
+ } catch (Exception e) {
+ logger.debug("Error disconnecting: {}", e.getMessage());
+ }
+ }
+
+ protected MQTTProtocolManager getProtocolManager() {
+ Acceptor acceptor = server.getRemotingService().getAcceptor(MQTT_PROTOCOL_NAME);
+ if (acceptor instanceof AbstractAcceptor abstractAcceptor) {
+ ProtocolManager protocolManager = abstractAcceptor.getProtocolMap().get(MQTT_PROTOCOL_NAME);
+ if (protocolManager instanceof MQTTProtocolManager mqttProtocolManager) {
+ return mqttProtocolManager;
+ }
+ }
+ return null;
+ }
+
+ protected DuplicateIDCache getPubCache(String clientId) {
+ return getCache(clientId, PacketIdCache.TYPE.PUBLISH);
+ }
+
+ protected int getPubCacheSize(String clientId) {
+ DuplicateIDCache cache = getPubCache(clientId);
+ return cache == null ? 0 : cache.getMap().size();
+ }
+
+ protected DuplicateIDCache getSubCache(String clientId) {
+ return getCache(clientId, PacketIdCache.TYPE.PUBREC);
+ }
+
+ protected int getSubCacheSize(String clientId) {
+ DuplicateIDCache cache = getSubCache(clientId);
+ return cache == null ? 0 : cache.getMap().size();
+ }
+
+ private DuplicateIDCache getCache(String clientId, PacketIdCache.TYPE type) {
+ SimpleString cacheName = PacketIdCache.getCacheName(server.getInternalNamingPrefix(), clientId, type);
+ if (server.getPostOffice().duplicateIDCacheExists(cacheName)) {
+ return server.getPostOffice().getDuplicateIDCache(cacheName);
+ }
+ return null;
+ }
+
+ protected Queue getSubscriptionQueue(String mqttTopicFilter, String clientId) {
+ return server.locateQueue(MQTTUtil.getCoreQueueFromMqttTopic(mqttTopicFilter, clientId, server.getConfiguration().getWildcardConfiguration()));
+ }
+
+ protected record BrokerRestartTask(ScheduledExecutorService scheduler, ScheduledFuture<?> future) {}
+
+ protected BrokerRestartTask startBrokerRestartTask(long delayMillis, Runnable restartLogger) {
+ ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
+ ScheduledFuture<?> future = scheduler.scheduleWithFixedDelay(() -> {
+ try {
+ if (restartLogger != null) {
+ restartLogger.run();
+ }
+ logger.info("Stopping broker");
+ server.stop();
+ waitForServerToStop(server);
+
+ logger.info("Compacting journal...");
+ compactJournal(server.getConfiguration().getJournalLocation(), server.getConfiguration().getJournalRetentionLocation(), ACTIVEMQ_DATA, "amq", server.getConfiguration().getJournalMinFiles(),
+ server.getConfiguration().getJournalPoolFiles(), server.getConfiguration().getJournalFileSize(), null, JournalRecordIds.UPDATE_DELIVERY_COUNT,
+ JournalRecordIds.SET_SCHEDULED_DELIVERY_TIME);
+ logger.info("Compacted journal.");
+
+ server.start();
+ waitForServerToStart(server);
+ } catch (Exception e) {
+ logger.warn("Error during broker restart", e);
+ }
+ }, delayMillis, delayMillis, TimeUnit.MILLISECONDS);
+ return new BrokerRestartTask(scheduler, future);
+ }
+
+ protected void stopBrokerRestartTask(BrokerRestartTask restartTask) throws Exception {
+ restartTask.future().cancel(true);
+ restartTask.scheduler().shutdownNow();
+ restartTask.scheduler().awaitTermination(10, TimeUnit.SECONDS);
+ if (!server.isStarted()) {
+ server.start();
+ waitForServerToStart(server);
+ }
+ }
+
+ protected record PublishResult(ExecutorService executor, Set<String> sentMessages, AtomicInteger publishErrors) {}
+
+ protected PublishResult startPublishing(List<Mqtt5BlockingClient> publishers, int numMessages) {
+ final ExecutorService publisherExecutor = Executors.newFixedThreadPool(publishers.size());
+ runAfter(() -> {
+ publisherExecutor.shutdownNow();
+ try {
+ publisherExecutor.awaitTermination(10, TimeUnit.SECONDS);
+ } catch (InterruptedException ignored) {
+ }
+ });
+ final Set<String> sentMessages = ConcurrentHashMap.newKeySet(publishers.size() * numMessages);
+ final AtomicInteger publishErrors = new AtomicInteger(0);
+ for (int i = 0; i < publishers.size(); i++) {
+ final int pubId = i;
+ final Mqtt5BlockingClient publisher = publishers.get(i);
+ publisherExecutor.execute(() -> {
+ for (int seq = 0; seq < numMessages; seq++) {
+ String payload = pubId + "-" + seq;
+ try {
+ waitForClientConnected(publisher);
+ Mqtt5PublishResult result = publisher.publishWith()
+ .topic(TOPIC)
+ .qos(MqttQos.EXACTLY_ONCE)
+ .payload(payload.getBytes(StandardCharsets.UTF_8))
+ .send();
+ if (result.getError().isPresent()) {
+ throw result.getError().get();
+ }
+ } catch (Throwable e) {
+ publishErrors.incrementAndGet();
+ logger.info("Pub failed: {}; in-flight QoS 2 state will be resumed on reconnect", payload, e);
+ }
+ sentMessages.add(payload);
+ }
+ logger.info("Publisher {} finished sending all {} messages", pubId, numMessages);
+ });
+ }
+ return new PublishResult(publisherExecutor, sentMessages, publishErrors);
+ }
+}
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
new file mode 100644
index 0000000000..f456d2a21f
--- /dev/null
+++ b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/mqtt/resiliency/QoS2SubscriberResiliencySoakTest.java
@@ -0,0 +1,184 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.tests.soak.mqtt.resiliency;
+
+import java.lang.invoke.MethodHandles;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
+import com.hivemq.client.mqtt.datatypes.MqttQos;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
+import org.apache.activemq.artemis.utils.TestParameters;
+import org.apache.activemq.artemis.utils.Wait;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+public class QoS2SubscriberResiliencySoakTest extends QoS2ResiliencySoakTestSupport {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static final String TEST_NAME = "QOS2_SUBSCRIBER_RESILIENCY_SOAK";
+
+ private static final int NUM_SUBSCRIBERS = TestParameters.testProperty(TEST_NAME, "NUM_SUBSCRIBERS", 5);
+ private static final int NUM_MESSAGES = TestParameters.testProperty(TEST_NAME, "NUM_MESSAGES", 10_000);
+ private static final int RESTART_PAUSE = TestParameters.testProperty(TEST_NAME, "RESTART_PAUSE", 2_000);
+ private static final int TIMEOUT_SECONDS = TestParameters.testProperty(TEST_NAME, "TIMEOUT_SECONDS", 180);
+
+ @Test
+ @Timeout(value = 5, unit = TimeUnit.MINUTES)
+ public void testQoS2SubscriberResiliency() throws Exception {
+ disableProtocolLogging();
+ logger.info("{} subscribers, {} messages/subscriber", NUM_SUBSCRIBERS, NUM_MESSAGES);
+
+ // create and subscribe then disconnect to leave idle subscriptions on the broker
+ final List<Mqtt5BlockingClient> subscribers = new ArrayList<>(NUM_SUBSCRIBERS);
+ runAfter(() -> subscribers.forEach(c -> {
+ try {
+ c.disconnect();
+ } catch (Exception ignored) {
+ }
+ }));
+ for (int i = 0; i < NUM_SUBSCRIBERS; i++) {
+ String clientId = "sub-" + i;
+ Mqtt5BlockingClient subscriber = createHiveMQClient(clientId, true);
+ subscriber.connectWith()
+ .cleanStart(false)
+ .sessionExpiryInterval(300)
+ .send();
+ subscriber.subscribeWith()
+ .topicFilter(TOPIC)
+ .qos(MqttQos.EXACTLY_ONCE)
+ .send();
+ subscriber.disconnect();
+ subscribers.add(subscriber);
+ assertNotNull(getSubscriptionQueue(TOPIC, getClientId(subscriber)));
+ }
+
+ // send messages using QoS 2
+ final Set<String> sentMessages = new HashSet<>(NUM_MESSAGES);
+ Mqtt5BlockingClient publisher = createHiveMQClient("pub", false);
+ publisher.connectWith().cleanStart(true).send();
+ logger.info("Publishing {} messages...", NUM_MESSAGES);
+ for (int seq = 0; seq < NUM_MESSAGES; seq++) {
+ String payload = String.valueOf(seq);
+ sentMessages.add(payload);
+ publisher.publishWith()
+ .topic(TOPIC)
+ .qos(MqttQos.EXACTLY_ONCE)
+ .payload(payload.getBytes(StandardCharsets.UTF_8))
+ .send();
+ }
+ logger.info("Published {} messages.", NUM_MESSAGES);
+ cleanDisconnect(publisher);
+
+ for (Mqtt5BlockingClient subscriber : subscribers) {
+ assertEquals(NUM_MESSAGES, getSubscriptionQueue(TOPIC, getClientId(subscriber)).getMessageCount());
+ }
+
+ // enableProtocolLogging();
+
+ final Map<String, Set<String>> receivedPerSubscriber = new HashMap<>();
+ final Map<String, Set<String>> duplicatesPerSubscriber = new HashMap<>();
+ final AtomicLong lastReceiveTime = new AtomicLong(System.currentTimeMillis());
+
+ for (Mqtt5BlockingClient subscriber : subscribers) {
+ final String clientId = getClientId(subscriber);
+ final Set<String> received = ConcurrentHashMap.newKeySet(NUM_MESSAGES);
+ receivedPerSubscriber.put(clientId, received);
+ final Set<String> duplicates = ConcurrentHashMap.newKeySet();
+ duplicatesPerSubscriber.put(clientId, duplicates);
+ subscriber.toAsync().publishes(MqttGlobalPublishFilter.ALL, publish -> {
+ String payload = new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8);
+ if (!received.add(payload)) {
+ logger.warn("Subscriber {} received duplicate: {}", clientId, payload);
+ duplicates.add(payload);
+ }
+ lastReceiveTime.set(System.currentTimeMillis());
+ });
+ subscriber.connectWith()
+ .cleanStart(false)
+ .sessionExpiryInterval(300)
+ .send();
+ logger.info("Subscriber {} reconnected", clientId);
+ }
+
+ // start broker restart task
+ BrokerRestartTask restartTask = startBrokerRestartTask(RESTART_PAUSE, () -> {
+ logger.info("===========");
+ for (Map.Entry<String, Set<String>> entry : receivedPerSubscriber.entrySet()) {
+ logger.info("Subscriber {} received {}/{} messages", entry.getKey(), entry.getValue().size(), NUM_MESSAGES);
+ }
+ logger.info("Last message received {}ms ago.", System.currentTimeMillis() - lastReceiveTime.get());
+ logger.info("===========");
+ });
+
+ final long SUBSCRIBER_TIMEOUT = 20_000;
+ Wait.assertTrue(() -> {
+ // quit early if subscribers are dead/stalled for some reason
+ if (System.currentTimeMillis() - lastReceiveTime.get() > SUBSCRIBER_TIMEOUT) {
+ for (Map.Entry<String, Set<String>> entry : receivedPerSubscriber.entrySet()) {
+ logger.warn("Subscriber {} received {}/{} messages", entry.getKey(), entry.getValue().size(), NUM_MESSAGES);
+ }
+ throw new AssertionError("No subscriber has received a message in " + SUBSCRIBER_TIMEOUT / 1000 + " seconds");
+ }
+ // any duplicate is a failure, no need to wait until the end
+ for (Map.Entry<String, Set<String>> duplicates : duplicatesPerSubscriber.entrySet()) {
+ assertEquals(0, duplicates.getValue().size(), "Subscriber " + duplicates.getKey() + " received duplicates: " + duplicates.getValue());
+ }
+ for (Set<String> received : receivedPerSubscriber.values()) {
+ if (received.size() < NUM_MESSAGES) {
+ return false;
+ }
+ }
+ return true;
+ }, TIMEOUT_SECONDS * 1000L, 100);
+
+ disableProtocolLogging();
+
+ // stop reconnection task, ensure broker is running
+ stopBrokerRestartTask(restartTask);
+
+ // enableProtocolLogging();
+
+ // 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)));
+ cleanDisconnect(subscriber);
+ assertNull(getSubCache(getClientId(subscriber)), "Sub cache should be null after clean start for " + getClientId(subscriber));
+ }
+ }
+}
\ 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