From 5bec488319a3f1a9adf6e4035ef46835db479797 Mon Sep 17 00:00:00 2001 From: borisalekseev Date: Sun, 16 Aug 2026 17:23:25 +0300 Subject: [PATCH] ARTEMIS-6190 fix MQTT Will message properties propagation All will properties are delivered with publish packet on unexpected client disconnect. --- .../protocol/mqtt/MQTTConnectionManager.java | 19 ++++---- .../core/protocol/mqtt/MQTTSession.java | 19 ++++---- .../core/protocol/mqtt/MQTTSessionState.java | 11 +++-- .../tests/integration/mqtt5/MQTT5Test.java | 46 +++++++++++++++---- 4 files changed, 62 insertions(+), 33 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 1a4031262d4f..83cd6de656d5 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 @@ -16,8 +16,6 @@ */ package org.apache.activemq.artemis.core.protocol.mqtt; -import java.util.List; - import io.netty.buffer.ByteBufAllocator; import io.netty.handler.codec.mqtt.MqttConnectMessage; import io.netty.handler.codec.mqtt.MqttProperties; @@ -93,18 +91,21 @@ synchronized void connect(MqttConnectMessage connect, String validatedUser, Stri session.getState().setWillRetain(connect.variableHeader().isWillRetain()); session.getState().setWillTopic(connect.payload().willTopic()); session.getState().setWillStatus(MQTTSessionState.WillStatus.NOT_SENT); + session.getState().setWillDelayInterval(0); + session.getState().setWillPublishProperties(MqttProperties.NO_PROPERTIES); 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()); - } - List userProperties = willProperties.getProperties(MqttProperties.MqttPropertyType.USER_PROPERTY.value()); - if (userProperties != null) { - session.getState().setWillUserProperties(userProperties); + MqttProperties publishProperties = new MqttProperties(); + for (MqttProperties.MqttProperty property : willProperties.listAll()) { + if (property.propertyId() == WILL_DELAY_INTERVAL.value()) { + session.getState().setWillDelayInterval((int) property.value()); + } else { + publishProperties.add(property); + } } + session.getState().setWillPublishProperties(publishProperties); } } } diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSession.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSession.java index 551ab416262d..80192bb7d944 100644 --- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSession.java +++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSession.java @@ -21,7 +21,6 @@ import io.netty.buffer.EmptyByteBuf; import io.netty.buffer.PooledByteBufAllocator; import io.netty.handler.codec.mqtt.MqttMessageBuilders; -import io.netty.handler.codec.mqtt.MqttProperties; import io.netty.handler.codec.mqtt.MqttPublishMessage; import io.netty.handler.codec.mqtt.MqttQoS; import org.apache.activemq.artemis.api.core.ActiveMQSecurityException; @@ -271,16 +270,14 @@ public void sendWillMessage() { if (state.getWillStatus() == MQTTSessionState.WillStatus.NOT_SENT) { try { state.setWillStatus(MQTTSessionState.WillStatus.SENDING); - MqttProperties properties; - if (state.getWillUserProperties() == null) { - properties = MqttProperties.NO_PROPERTIES; - } else { - properties = new MqttProperties(); - for (MqttProperties.MqttProperty userProperty : state.getWillUserProperties()) { - properties.add(userProperty); - } - } - MqttPublishMessage publishMessage = MqttMessageBuilders.publish().messageId(0).qos(MqttQoS.valueOf(state.getWillQoSLevel())).retained(state.isWillRetain()).topicName(state.getWillTopic()).payload(state.getWillMessage() == null ? new EmptyByteBuf(PooledByteBufAllocator.DEFAULT) : state.getWillMessage()).properties(properties).build(); + MqttPublishMessage publishMessage = MqttMessageBuilders + .publish() + .messageId(0) + .qos(MqttQoS.valueOf(state.getWillQoSLevel())) + .retained(state.isWillRetain()) + .topicName(state.getWillTopic()) + .payload(state.getWillMessage() == null ? new EmptyByteBuf(PooledByteBufAllocator.DEFAULT) : state.getWillMessage()) + .properties(state.getWillPublishProperties()).build(); logger.debug("{} sending will message: {}", this, publishMessage); getMqttPublishManager().sendToQueue(publishMessage, true); state.setWillStatus(MQTTSessionState.WillStatus.SENT); diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionState.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionState.java index 8397abee1678..c0304a07c633 100644 --- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionState.java +++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionState.java @@ -83,7 +83,7 @@ public class MQTTSessionState { private long willDelayInterval = 0; - private List willUserProperties; + private MqttProperties willPublishProperties = MqttProperties.NO_PROPERTIES; private WillStatus willStatus = WillStatus.NOT_SENT; @@ -163,6 +163,7 @@ public synchronized void clear() throws Exception { willStatus = WillStatus.NOT_SENT; failed = false; willDelayInterval = 0; + willPublishProperties = MqttProperties.NO_PROPERTIES; willRetain = false; willTopic = null; clientMaxPacketSize = 0; @@ -318,12 +319,12 @@ public void setWillDelayInterval(long willDelayInterval) { this.willDelayInterval = willDelayInterval; } - public void setWillUserProperties(List userProperties) { - this.willUserProperties = userProperties; + public void setWillPublishProperties(MqttProperties willPublishProperties) { + this.willPublishProperties = willPublishProperties; } - public List getWillUserProperties() { - return willUserProperties; + public MqttProperties getWillPublishProperties() { + return willPublishProperties; } public WillStatus getWillStatus() { diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5Test.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5Test.java index dcae8935a29b..cf9bcd8603c2 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5Test.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5Test.java @@ -68,6 +68,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -272,13 +273,16 @@ public void testAddressAutoCreationNegative() throws Exception { } /** - * There is no normative statement in the spec about supporting user properties on will messages, but it is implied - * in various places. + * Verifies that the Application Message properties in Will Properties are included in the published Will Message. */ @Test @Timeout(DEFAULT_TIMEOUT_SEC) public void testWillMessageProperties() throws Exception { final byte[] WILL = RandomUtil.randomBytes(); + final long MESSAGE_EXPIRY_INTERVAL = 60; + final String CONTENT_TYPE = "application/octet-stream"; + final String RESPONSE_TOPIC = "/topic/reply"; + final byte[] CORRELATION_DATA = RandomUtil.randomBytes(); final String[][] properties = new String[10][2]; for (String[] property : properties) { property[0] = RandomUtil.randomUUIDString(); @@ -287,16 +291,18 @@ public void testWillMessageProperties() throws Exception { // consumer of the will message MqttClient client1 = createPahoClient("willConsumer"); + runAfter(() -> { + if (client1.isConnected()) { + client1.disconnect(); + } + client1.close(); + }); CountDownLatch latch = new CountDownLatch(1); + AtomicReference receivedProperties = new AtomicReference<>(); client1.setCallback(new DefaultMqttCallback() { @Override public void messageArrived(String topic, MqttMessage message) { - int i = 0; - for (UserProperty property : message.getProperties().getUserProperties()) { - assertEquals(properties[i][0], property.getKey()); - assertEquals(properties[i][1], property.getValue()); - i++; - } + receivedProperties.set(message.getProperties()); latch.countDown(); } }); @@ -305,7 +311,18 @@ public void messageArrived(String topic, MqttMessage message) { // consumer to generate the will MqttClient client2 = createPahoClient("willGenerator"); + runAfter(() -> { + if (client2.isConnected()) { + client2.disconnectForcibly(0, 0, false); + } + client2.close(); + }); MqttProperties willMessageProperties = new MqttProperties(); + willMessageProperties.setPayloadFormat(true); + willMessageProperties.setMessageExpiryInterval(MESSAGE_EXPIRY_INTERVAL); + willMessageProperties.setContentType(CONTENT_TYPE); + willMessageProperties.setResponseTopic(RESPONSE_TOPIC); + willMessageProperties.setCorrelationData(CORRELATION_DATA); List userProperties = new ArrayList<>(); for (String[] property : properties) { userProperties.add(new UserProperty(property[0], property[1])); @@ -318,6 +335,19 @@ public void messageArrived(String topic, MqttMessage message) { client2.connect(options); client2.disconnectForcibly(0, 0, false); assertTrue(latch.await(2, TimeUnit.SECONDS)); + + MqttProperties actualProperties = receivedProperties.get(); + assertTrue(actualProperties.getPayloadFormat()); + assertTrue(actualProperties.getMessageExpiryInterval() > 0); + assertTrue(actualProperties.getMessageExpiryInterval() <= MESSAGE_EXPIRY_INTERVAL); + assertEquals(CONTENT_TYPE, actualProperties.getContentType()); + assertEquals(RESPONSE_TOPIC, actualProperties.getResponseTopic()); + assertArrayEquals(CORRELATION_DATA, actualProperties.getCorrelationData()); + assertEquals(properties.length, actualProperties.getUserProperties().size()); + for (int i = 0; i < properties.length; i++) { + assertEquals(properties[i][0], actualProperties.getUserProperties().get(i).getKey()); + assertEquals(properties[i][1], actualProperties.getUserProperties().get(i).getValue()); + } } /**