This is an automated email from the ASF dual-hosted git repository.
tabish pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/main by this push:
new 4362b39840 ARTEMIS-5522 MQTT LWT not sent after subsequent disconnects
on same session
4362b39840 is described below
commit 4362b39840675a1955736cf7949646f729837ac0
Author: Justin Bertram <[email protected]>
AuthorDate: Thu Jun 5 12:07:47 2025 -0500
ARTEMIS-5522 MQTT LWT not sent after subsequent disconnects on same session
---
.../core/protocol/mqtt/MQTTConnectionManager.java | 3 ++-
.../mqtt5/spec/controlpackets/ConnectTests.java | 27 ++++++++++++++--------
2 files changed, 20 insertions(+), 10 deletions(-)
diff --git
a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnectionManager.java
b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnectionManager.java
index a4a12cb22a..ebc9dd6e5a 100644
---
a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnectionManager.java
+++
b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnectionManager.java
@@ -95,13 +95,14 @@ public class MQTTConnectionManager {
session.getState().setWillQoSLevel(connect.variableHeader().willQos());
session.getState().setWillRetain(connect.variableHeader().isWillRetain());
session.getState().setWillTopic(connect.payload().willTopic());
+
session.getState().setWillStatus(MQTTSessionState.WillStatus.NOT_SENT);
if (session.getVersion() == MQTTVersion.MQTT_5) {
MqttProperties willProperties = connect.payload().willProperties();
if (willProperties != null) {
MqttProperties.MqttProperty willDelayInterval =
willProperties.getProperty(WILL_DELAY_INTERVAL.value());
if (willDelayInterval != null) {
- session.getState().setWillDelayInterval(( int)
willDelayInterval.value());
+ session.getState().setWillDelayInterval((int)
willDelayInterval.value());
}
List<? extends MqttProperties.MqttProperty> userProperties =
willProperties.getProperties(MqttProperties.MqttPropertyType.USER_PROPERTY.value());
if (userProperties != null) {
diff --git
a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/spec/controlpackets/ConnectTests.java
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/spec/controlpackets/ConnectTests.java
index 4d5788fdd9..d236a56d88 100644
---
a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/spec/controlpackets/ConnectTests.java
+++
b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/spec/controlpackets/ConnectTests.java
@@ -365,8 +365,6 @@ public class ConnectTests extends MQTT5TestSupport {
// consumer of the will message
MqttClient willConsumer = createPahoClient(WILL_CONSUMER);
- CountDownLatch latch = new CountDownLatch(1);
- willConsumer.setCallback(new LatchedMqttCallback(latch));
willConsumer.connect();
willConsumer.subscribe("/topic/foo", 1);
@@ -377,16 +375,27 @@ public class ConnectTests extends MQTT5TestSupport {
MqttConnectionOptions options = new MqttConnectionOptionsBuilder()
.sessionExpiryInterval(5L)
.will("/topic/foo", new MqttMessage(WILL))
+ .cleanStart(false)
.build();
options.setWillMessageProperties(willMessageProperties);
- willSender.connect(options);
- MQTTSessionState state = getSessionStates().get(WILL_SENDER);
- assertNotNull(state);
- assertNotNull(state.getWillMessage());
- willSender.disconnectForcibly(0, 0, false);
- assertTrue(latch.await(2, TimeUnit.SECONDS));
- assertNull(state.getWillMessage());
+ // ensure the will message is sent & removed on each disconnect even
after reconnection to the same session
+ for (int i = 0; i < 5; i++) {
+ willSender.connect(options);
+ MQTTSessionState state = getSessionStates().get(WILL_SENDER);
+ assertNotNull(state);
+ assertNotNull(state.getWillMessage());
+ CountDownLatch latch = new CountDownLatch(1);
+ willConsumer.setCallback(new LatchedMqttCallback(latch));
+ willSender.disconnectForcibly(0, 0, false);
+
+ assertTrue(latch.await(2, TimeUnit.SECONDS));
+ assertNull(state.getWillMessage());
+ }
+
+ willSender.close();
+ willConsumer.disconnect();
+ willConsumer.close();
}
/*
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
