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:27 UTC
(artemis) 02/04: ARTEMIS-6189 implement MQTT QoS 1 & 2 resiliency
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 b30557853d73e73be1fa421f38e1b95d2c97762b
Author: Justin Bertram <jb...@apache.org>
AuthorDate: Wed Aug 12 14:37:55 2026 -0500
ARTEMIS-6189 implement MQTT QoS 1 & 2 resiliency
Replace the in-memory and per-client resources with journal-persisted
packet-ID correlation and DuplicateIDCache-backed packet ID caches
so that QoS 1 and QoS 2 message flows survive broker restarts.
- Packet ID assignments are persisted to the journal via a
JournalHashMap so that when a message is redelivered after a broker
restart the same packet ID is reused
- Inbound QoS 2 duplicate detection and outbound QoS 2 PUBREC tracking
use DuplicateIDCache instead of in-memory sets, surviving restarts
- The per-client QoS 2 management queue (used for PUBREL persistence)
is eliminated; PUBREL retransmission is driven by iterating the
PUBREC cache on reconnect
- Objects related to resilience (caches, journal maps) are only created
when actually used, avoiding overhead for QoS 0/1-only clients
- Sends and acknowledgements are now explicitly transactional, ensuring
atomicity between message operations and correlation store and cache
updates
- Session startup is asynchronous with CountDownLatch synchronization,
preventing a race where client packets arrive before the session is
fully initialized after CONNACK
- Duplicate QoS 2 publish detection is moved earlier into the protocol
handler
- MQTT message filters are constructed once in the protocol manager and
shared, rather than duplicated per subscription manager instance
- Each subscription item now directly holds its consumer reference,
eliminating a separate consumer lookup map
- Graceful handling of ActiveMQShutdownException during message deletion
and rollback redelivery avoids misleading log messages during broker
restart
- Error logging throughout the MQTT protocol includes client ID and
packet ID context for easier debugging
Co-Authored-By: Claude <no...@anthropic.com>
---
.../core/protocol/mqtt/CoreDeliveryInfo.java | 79 ++++
.../artemis/core/protocol/mqtt/MQTTBundle.java | 3 +
.../artemis/core/protocol/mqtt/MQTTConnection.java | 2 +-
.../core/protocol/mqtt/MQTTConnectionManager.java | 15 +-
.../artemis/core/protocol/mqtt/MQTTLogger.java | 29 +-
.../core/protocol/mqtt/MQTTMessageInfo.java | 52 ---
.../core/protocol/mqtt/MQTTProtocolHandler.java | 92 +++--
.../core/protocol/mqtt/MQTTProtocolManager.java | 75 +++-
.../protocol/mqtt/MQTTProtocolManagerFactory.java | 3 +-
.../core/protocol/mqtt/MQTTPublishManager.java | 400 ++++++++++-----------
.../artemis/core/protocol/mqtt/MQTTSession.java | 11 +-
.../core/protocol/mqtt/MQTTSessionCallback.java | 8 +-
.../core/protocol/mqtt/MQTTSessionState.java | 349 ++++++++----------
.../core/protocol/mqtt/MQTTStateManager.java | 120 ++++---
.../protocol/mqtt/MQTTSubscriptionManager.java | 170 ++++-----
.../artemis/core/protocol/mqtt/MQTTUtil.java | 226 ++++++------
.../artemis/core/protocol/mqtt/PacketIdCache.java | 116 ++++++
.../core/protocol/mqtt/PacketIdCorrelationKey.java | 133 +++++++
.../core/protocol/mqtt/SubscriptionItem.java | 83 +++++
.../core/protocol/mqtt/MQTTSessionStateTest.java | 14 +-
.../core/protocol/mqtt/MQTTStateManagerTest.java | 3 +
.../artemis/core/protocol/mqtt/StateSerDeTest.java | 6 +-
.../persistence/impl/journal/JournalRecordIds.java | 2 +
.../artemis/core/postoffice/DuplicateIDCache.java | 2 +-
.../artemis/core/postoffice/PostOffice.java | 4 +
.../postoffice/impl/InMemoryDuplicateIDCache.java | 8 +-
.../core/postoffice/impl/NoOpDuplicateIDCache.java | 4 +-
.../impl/PersistentDuplicateIDCache.java | 14 +-
.../core/postoffice/impl/PostOfficeImpl.java | 8 +-
.../core/remoting/impl/invm/InVMAcceptor.java | 2 +-
.../core/remoting/impl/netty/NettyAcceptor.java | 2 +-
.../core/remoting/server/RemotingService.java | 2 +
.../remoting/server/impl/RemotingServiceImpl.java | 13 +-
.../artemis/core/server/ActiveMQMessageBundle.java | 2 +-
.../artemis/core/server/ActiveMQServerLogger.java | 10 +-
.../core/server/impl/ActiveMQServerImpl.java | 6 +-
.../artemis/core/server/impl/QueueImpl.java | 3 +
.../artemis/core/server/impl/RefsOperation.java | 3 +
.../core/server/impl/ServerSessionImpl.java | 5 +-
.../protocol/AbstractProtocolManagerFactory.java | 2 +-
.../spi/core/protocol/ProtocolManagerFactory.java | 2 +-
docs/user-manual/versions.adoc | 12 +
.../core/server/impl/fakes/FakePostOffice.java | 9 +
.../artemis/tests/integration/mqtt/MQTTTest.java | 26 --
.../integration/mqtt/PahoMQTTQOS2SecurityTest.java | 12 +-
.../artemis/tests/integration/mqtt5/MQTT5Test.java | 63 +++-
.../tests/integration/mqtt5/MQTT5TestSupport.java | 80 +++++
.../mqtt5/spec/controlpackets/ConnAckTests.java | 11 +
.../mqtt5/spec/controlpackets/ConnectTests.java | 37 +-
49 files changed, 1501 insertions(+), 832 deletions(-)
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/CoreDeliveryInfo.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/CoreDeliveryInfo.java
new file mode 100644
index 0000000000..853210c740
--- /dev/null
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/CoreDeliveryInfo.java
@@ -0,0 +1,79 @@
+/*
+ * 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.core.protocol.mqtt;
+
+import java.util.Objects;
+
+import org.apache.activemq.artemis.api.core.SimpleString;
+
+/**
+ * Tracks volatile, in-memory state for an in-flight MQTT delivery: the consumer that originated it and the
+ * {@link PacketIdCorrelationKey} that identifies the underlying core message and address. This information is held only
+ * for the lifetime of the connection and is discarded on disconnect, whereas the {@link PacketIdCorrelationKey} mapping
+ * is persisted in the journal so that packet IDs can be correlated across reconnects.
+ */
+public class CoreDeliveryInfo {
+ private long consumerId;
+ private PacketIdCorrelationKey packetIdCorrelationKey;
+
+ public static CoreDeliveryInfo of(long consumerId, PacketIdCorrelationKey packetIdCorrelationKey) {
+ return new CoreDeliveryInfo(consumerId, packetIdCorrelationKey);
+ }
+
+ private CoreDeliveryInfo(long consumerId, PacketIdCorrelationKey packetIdCorrelationKey) {
+ this.consumerId = consumerId;
+ this.packetIdCorrelationKey = packetIdCorrelationKey;
+ }
+
+ public long getConsumerId() {
+ return consumerId;
+ }
+
+ public PacketIdCorrelationKey getPacketIdCorrelationKey() {
+ return packetIdCorrelationKey;
+ }
+
+ public long getCoreMessageId() {
+ return packetIdCorrelationKey.getCoreMessageId();
+ }
+
+ public SimpleString getAddress() {
+ return packetIdCorrelationKey.getAddress();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof CoreDeliveryInfo other)) {
+ return false;
+ }
+ return consumerId == other.consumerId &&
+ Objects.equals(packetIdCorrelationKey, other.packetIdCorrelationKey);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(consumerId, packetIdCorrelationKey);
+ }
+
+ @Override
+ public String toString() {
+ return "CoreDeliveryInfo[" + "consumerId=" + consumerId + ", packetIdCorrelationKey=" + packetIdCorrelationKey + "]";
+ }
+}
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTBundle.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTBundle.java
index 42e5d472e5..88563f636b 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTBundle.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTBundle.java
@@ -33,4 +33,7 @@ public interface MQTTBundle {
@Message(id = 850001, value = "Unable to generate MQTT packet ID. All valid values between 1 and 65535 are in use. IDs will become available as messages are acknowledged by the client that has received them.")
IllegalStateException unableToGenerateID();
+
+ @Message(id = 850002, value = "StorageManager is null")
+ IllegalStateException storageManagerIsNull();
}
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnection.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnection.java
index 5184f33468..55ffe6a181 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnection.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnection.java
@@ -40,7 +40,7 @@ public class MQTTConnection extends AbstractRemotingConnection {
private boolean clientIdAssignedByBroker = false;
- public MQTTConnection(Connection transportConnection) throws Exception {
+ public MQTTConnection(Connection transportConnection) {
super(transportConnection, null);
this.destroyed = false;
}
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 1a4031262d..40194d2de8 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
@@ -123,10 +123,21 @@ public class MQTTConnectionManager {
connackProperties = MqttProperties.NO_PROPERTIES;
}
+ if (sessionState.getClientSessionExpiryInterval() == -1 || sessionState.getClientSessionExpiryInterval() > 0) {
+ session.getStateManager().storeDurableState(sessionState);
+ }
+
session.getConnection().setConnected(true);
session.getProtocolHandler().sendConnack(MQTTReasonCodes.SUCCESS, sessionPresent && !cleanStart, connackProperties);
- // ensure we don't publish before the CONNACK
- session.start();
+ // [MQTT-3.2.0-1] the CONNACK is sent via IO callback so the session should be started the same way to avoid a race
+ session.getProtocolHandler().runAfterStorageOperations(() -> {
+ try {
+ session.start();
+ } catch (Exception e) {
+ MQTTLogger.LOGGER.errorDisconnectingClient(e);
+ disconnect(true);
+ }
+ });
}
private MqttProperties getConnackProperties() {
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTLogger.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTLogger.java
index 7a2ef08b2c..b39baa360e 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTLogger.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTLogger.java
@@ -29,11 +29,11 @@ public interface MQTTLogger {
MQTTLogger LOGGER = BundleFactory.newBundle(MQTTLogger.class, MQTTLogger.class.getPackage().getName());
- @LogMessage(id = 832000, value = "Unable to send message: {}", level = LogMessage.Level.WARN)
- void unableToSendMessage(MessageReference message, Exception e);
+ @LogMessage(id = 832000, value = "Unable to send message to MQTT client {}: {}", level = LogMessage.Level.WARN)
+ void unableToSendMessage(String clientId, MessageReference message, Exception e);
- @LogMessage(id = 832001, value = "MQTT client({}) attempted to ack already ack'd message: ", level = LogMessage.Level.WARN)
- void failedToAckMessage(String clientId, Exception e);
+ @LogMessage(id = 832001, value = "MQTT client {} failed to acknowledge message: {}", level = LogMessage.Level.WARN)
+ void failedToAckMessage(String clientId, String exceptionMessage);
@LogMessage(id = 834000, value = "Error removing subscription.", level = LogMessage.Level.ERROR)
void errorRemovingSubscription(Exception e);
@@ -41,8 +41,8 @@ public interface MQTTLogger {
@LogMessage(id = 834001, value = "Error disconnecting client.", level = LogMessage.Level.ERROR)
void errorDisconnectingClient(Exception e);
- @LogMessage(id = 834002, value = "Error processing control packet: {}", level = LogMessage.Level.ERROR)
- void errorProcessingControlPacket(String packet, Exception e);
+ @LogMessage(id = 834002, value = "Error processing MQTT packet; client ID: {}; packet: {}; {}", level = LogMessage.Level.ERROR)
+ void errorProcessingPacket(String clientId, String packet, String exceptionMessage, Exception e);
@LogMessage(id = 834003, value = "Error sending will message.", level = LogMessage.Level.ERROR)
void errorSendingWillMessage(Exception e);
@@ -53,8 +53,8 @@ public interface MQTTLogger {
@LogMessage(id = 834005, value = "Failed to cast property {}.", level = LogMessage.Level.ERROR)
void failedToCastProperty(String property);
- @LogMessage(id = 834006, value = "Failed to publish MQTT message: {}.", level = LogMessage.Level.ERROR)
- void failedToPublishMqttMessage(String exceptionMessage, Throwable t);
+ @LogMessage(id = 834006, value = "Failed to publish MQTT message. Client ID: {}. Packet ID: {}. Exception message: {}", level = LogMessage.Level.ERROR)
+ void failedToPublishMqttMessage(String clientId, int packetId, String exceptionMessage, Throwable t);
@LogMessage(id = 834007, value = "Authorization failure sending will message: {}", level = LogMessage.Level.ERROR)
void authorizationFailureSendingWillMessage(String message);
@@ -62,8 +62,8 @@ public interface MQTTLogger {
@LogMessage(id = 834008, value = "Failed to remove session state for client with ID: {}", level = LogMessage.Level.ERROR)
void failedToRemoveSessionState(String clientID, Exception e);
- @LogMessage(id = 834009, value = "Ignoring duplicate MQTT QoS2 PUBLISH packet for packet ID {} from client with ID {}.", level = LogMessage.Level.WARN)
- void ignoringQoS2Publish(String clientId, long packetId);
+ @LogMessage(id = 834009, value = "Ignoring duplicate MQTT QoS2 PUBLISH; packet ID: {}; client ID: {}.", level = LogMessage.Level.WARN)
+ void ignoringQoS2Publish(long packetId, String clientId);
@LogMessage(id = 834010, value = "Unable to scan MQTT sessions", level = LogMessage.Level.ERROR)
void unableToScanSessions(Exception e);
@@ -76,4 +76,13 @@ public interface MQTTLogger {
@LogMessage(id = 834013, value = "Invalid MQTT session state message. Will not load this state into memory.", level = LogMessage.Level.WARN)
void errorDeserializingStateMessage(Exception e);
+
+ @LogMessage(id = 834014, value = "MQTT client {} sent PUBREC for packet {}, but acknowledgement failed. Internal consumer {} not found. Internal session is {}.", level = LogMessage.Level.WARN)
+ void failedToAckMessageConsumerNotFound(String clientId, int packetId, long consumerId, String closed);
+
+ @LogMessage(id = 834015, value = "Unable to handle MQTT packet [{}] from {}. Internal session is closed.", level = LogMessage.Level.ERROR)
+ void internalSessionClosed(String packet, String clientId);
+
+ @LogMessage(id = 834016, value = "Storage operation failed. Error code: {}; message: {}", level = LogMessage.Level.ERROR)
+ void storageOperationError(int errorCode, String errorMessage);
}
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTMessageInfo.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTMessageInfo.java
deleted file mode 100644
index f25640008e..0000000000
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTMessageInfo.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * 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.core.protocol.mqtt;
-
-/**
- * MQTT Acks only hold message ID information. From this we must infer the internal message ID and consumer.
- */
-class MQTTMessageInfo {
-
- private long serverMessageId;
-
- private long consumerId;
-
- private String address;
-
- MQTTMessageInfo(long serverMessageId, long consumerId, String address) {
- this.serverMessageId = serverMessageId;
- this.consumerId = consumerId;
- this.address = address;
- }
-
- long getServerMessageId() {
- return serverMessageId;
- }
-
- long getConsumerId() {
- return consumerId;
- }
-
- String getAddress() {
- return address;
- }
-
- @Override
- public String toString() {
- return ("ServerMessageId: " + serverMessageId + " ConsumerId: " + consumerId + " addr: " + address);
- }
-}
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolHandler.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolHandler.java
index 071df5c5fb..5673742e0e 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolHandler.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolHandler.java
@@ -45,6 +45,8 @@ import io.netty.util.ReferenceCountUtil;
import org.apache.activemq.artemis.api.core.ActiveMQSecurityException;
import org.apache.activemq.artemis.api.core.Pair;
import org.apache.activemq.artemis.core.io.IOCallback;
+import org.apache.activemq.artemis.core.persistence.OperationContext;
+import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.protocol.mqtt.exceptions.DisconnectException;
import org.apache.activemq.artemis.core.protocol.mqtt.exceptions.InvalidClientIdException;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
@@ -91,11 +93,10 @@ public class MQTTProtocolHandler extends ChannelInboundHandlerAdapter {
this.mqttMessageActor = new Actor<>(server.getThreadPool(), this::act);
}
- void setConnection(MQTTConnection connection, ConnectionEntry entry) throws Exception {
+ void setConnection(MQTTConnection connection, ConnectionEntry entry) {
this.connectionEntry = entry;
this.connection = connection;
this.session = new MQTTSession(this, connection, protocolManager, server.getConfiguration().getWildcardConfiguration(), server.newOperationContext());
- server.getStorageManager().setContext(session.getSessionContext());
}
void stop() {
@@ -124,6 +125,15 @@ public class MQTTProtocolHandler extends ChannelInboundHandlerAdapter {
return;
}
+ if (session.getServerSession() != null && session.getServerSession().isClosed()) {
+ MQTTLogger.LOGGER.internalSessionClosed(MQTTUtil.getMessageForLogging(message, session.getVersion()), session.getState().getClientId());
+ if (session.getVersion() == MQTTVersion.MQTT_5) {
+ sendDisconnect(MQTTReasonCodes.IMPLEMENTATION_SPECIFIC_ERROR);
+ }
+ disconnect(true);
+ return;
+ }
+
String interceptResult = this.protocolManager.invokeIncoming(message, this.connection);
if (interceptResult != null) {
logger.debug("Interceptor {} rejected MQTT control packet: {}", interceptResult, message);
@@ -151,7 +161,18 @@ public class MQTTProtocolHandler extends ChannelInboundHandlerAdapter {
}
}
+ private OperationContext recoverContext() {
+ OperationContext oldContext = server.getStorageManager().getContext();
+ server.getStorageManager().setContext(session.getSessionContext());
+ return oldContext;
+ }
+
+ private void resetContext(OperationContext oldContext) {
+ server.getStorageManager().setContext(oldContext);
+ }
+
public void act(MqttMessage message) {
+ OperationContext oldContext = recoverContext();
try {
switch (message.fixedHeader().messageType()) {
case AUTH:
@@ -193,13 +214,14 @@ public class MQTTProtocolHandler extends ChannelInboundHandlerAdapter {
disconnect(true);
}
} catch (Exception e) {
- MQTTLogger.LOGGER.errorProcessingControlPacket(message.toString(), e);
+ MQTTLogger.LOGGER.errorProcessingPacket(session.getState().getClientId(), MQTTUtil.getMessageForLogging(message, session.getVersion()), e.getMessage(), e);
if (session.getVersion() == MQTTVersion.MQTT_5) {
sendDisconnect(MQTTReasonCodes.IMPLEMENTATION_SPECIFIC_ERROR);
}
disconnect(true);
} finally {
ReferenceCountUtil.release(message);
+ resetContext(oldContext);
}
}
@@ -273,8 +295,8 @@ public class MQTTProtocolHandler extends ChannelInboundHandlerAdapter {
}
void disconnect(boolean error, MqttMessage disconnect) {
- if (disconnect != null && disconnect.variableHeader() instanceof MqttReasonCodeAndPropertiesVariableHeader) {
- Integer sessionExpiryInterval = MQTTUtil.getProperty(Integer.class, ((MqttReasonCodeAndPropertiesVariableHeader)disconnect.variableHeader()).properties(), SESSION_EXPIRY_INTERVAL, null);
+ if (disconnect != null && disconnect.variableHeader() instanceof MqttReasonCodeAndPropertiesVariableHeader variableHeader) {
+ Integer sessionExpiryInterval = MQTTUtil.getProperty(Integer.class, variableHeader.properties(), SESSION_EXPIRY_INTERVAL, null);
if (sessionExpiryInterval != null) {
session.getState().setClientSessionExpiryInterval(sessionExpiryInterval);
}
@@ -318,6 +340,12 @@ public class MQTTProtocolHandler extends ChannelInboundHandlerAdapter {
return;
}
+ if (message.fixedHeader().qosLevel().value() == 2 && session.getState().getPublishCache().contains(message.variableHeader().packetId())) {
+ MQTTLogger.LOGGER.ignoringQoS2Publish(message.variableHeader().packetId(), session.getState().getClientId());
+ sendPubRec(message.variableHeader().packetId(), MQTTReasonCodes.SUCCESS);
+ return;
+ }
+
try {
session.getMqttPublishManager().sendToQueue(message, false);
} catch (DisconnectException e) {
@@ -332,8 +360,8 @@ public class MQTTProtocolHandler extends ChannelInboundHandlerAdapter {
sendPublishProtocolControlMessage(messageId, MqttMessageType.PUBACK, reasonCode);
}
- void sendPubRel(int messageId) {
- sendPublishProtocolControlMessage(messageId, MqttMessageType.PUBREL);
+ void sendPubRel(int messageId, byte reasonCode) {
+ sendPublishProtocolControlMessage(messageId, MqttMessageType.PUBREL, reasonCode);
}
void sendPubRec(int messageId, byte reasonCode) {
@@ -341,11 +369,7 @@ public class MQTTProtocolHandler extends ChannelInboundHandlerAdapter {
}
void sendPubComp(int messageId) {
- sendPublishProtocolControlMessage(messageId, MqttMessageType.PUBCOMP);
- }
-
- void sendPublishProtocolControlMessage(int messageId, MqttMessageType messageType) {
- sendPublishProtocolControlMessage(messageId, messageType, MQTTReasonCodes.SUCCESS);
+ sendPublishProtocolControlMessage(messageId, MqttMessageType.PUBCOMP, MQTTReasonCodes.SUCCESS);
}
void sendPublishProtocolControlMessage(int messageId, MqttMessageType messageType, byte reasonCode) {
@@ -364,19 +388,27 @@ public class MQTTProtocolHandler extends ChannelInboundHandlerAdapter {
void handlePuback(MqttPubAckMessage message) throws Exception {
// ((MqttPubReplyMessageVariableHeader)message.variableHeader()).reasonCode();
- session.getMqttPublishManager().handlePubAck(getMessageId(message));
+ session.getMqttPublishManager().handlePubAck(getPacketId(message));
}
void handlePubrec(MqttMessage message) throws Exception {
- session.getMqttPublishManager().handlePubRec(getMessageId(message));
+ byte reasonCode = MQTTReasonCodes.SUCCESS;
+ if (message.variableHeader() instanceof MqttPubReplyMessageVariableHeader header) {
+ reasonCode = header.reasonCode();
+ }
+ if ((reasonCode & 0xFF) >= 0x80) {
+ session.getMqttPublishManager().handlePubRecError(getPacketId(message));
+ } else {
+ session.getMqttPublishManager().handlePubRec(getPacketId(message));
+ }
}
- void handlePubrel(MqttMessage message) {
- session.getMqttPublishManager().handlePubRel(getMessageId(message));
+ void handlePubrel(MqttMessage message) throws Exception {
+ session.getMqttPublishManager().handlePubRel(getPacketId(message));
}
void handlePubcomp(MqttMessage message) throws Exception {
- session.getMqttPublishManager().handlePubComp(getMessageId(message));
+ session.getMqttPublishManager().handlePubComp(getPacketId(message));
}
void handleSubscribe(MqttSubscribeMessage message) throws Exception {
@@ -410,19 +442,28 @@ public class MQTTProtocolHandler extends ChannelInboundHandlerAdapter {
return;
}
MQTTUtil.logMessage(session.getState(), message, false, session.getVersion());
- server.getStorageManager().afterCompleteOperations(new IOCallback() {
+ runAfterStorageOperations(() -> ctx.writeAndFlush(message, ctx.voidPromise()));
+ }
+
+ void runAfterStorageOperations(Runnable runnable) {
+ StorageManager storageManager = server.getStorageManager();
+ if (storageManager == null) {
+ throw MQTTBundle.BUNDLE.storageManagerIsNull();
+ }
+ storageManager.afterCompleteOperations(new IOCallback() {
@Override
public void done() {
- ctx.writeAndFlush(message, ctx.voidPromise());
+ runnable.run();
}
@Override
public void onError(int errorCode, String errorMessage) {
+ MQTTLogger.LOGGER.storageOperationError(errorCode, errorMessage);
}
});
}
- private int getMessageId(MqttMessage message) {
+ private int getPacketId(MqttMessage message) {
return ((MqttMessageIdVariableHeader) message.variableHeader()).messageId();
}
@@ -528,16 +569,7 @@ public class MQTTProtocolHandler extends ChannelInboundHandlerAdapter {
session.getProtocolHandler().sendConnack(MQTTReasonCodes.NOT_AUTHORIZED_3);
}
// avoid a race with sending the CONNACK packet and disconnecting the client
- server.getStorageManager().afterCompleteOperations(new IOCallback() {
- @Override
- public void done() {
- disconnect(true);
- }
-
- @Override
- public void onError(int errorCode, String errorMessage) {
- }
- });
+ runAfterStorageOperations(() -> disconnect(true));
result = Boolean.FALSE;
}
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManager.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManager.java
index 575efccbde..2161c4754f 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManager.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManager.java
@@ -29,6 +29,7 @@ import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.codec.mqtt.MqttMessage;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.BaseInterceptor;
+import org.apache.activemq.artemis.api.core.FilterConstants;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.management.CoreNotificationType;
import org.apache.activemq.artemis.api.core.management.ManagementHelper;
@@ -76,6 +77,10 @@ public class MQTTProtocolManager extends AbstractProtocolManager<MqttMessage, MQ
private final MQTTRoutingHandler routingHandler;
+ private final SimpleString messageFilter;
+
+ private final SimpleString messageFilterNoDollar;
+
private MQTTStateManager sessionStateManager;
MQTTProtocolManager(ActiveMQServer server,
@@ -84,8 +89,14 @@ public class MQTTProtocolManager extends AbstractProtocolManager<MqttMessage, MQ
this.server = server;
this.updateInterceptors(incomingInterceptors, outgoingInterceptors);
server.getManagementService().addNotificationListener(this);
+
+ StringBuilder baseFilter = getBaseFilter(server);
+ this.messageFilter = getMessageFilter(baseFilter);
+ this.messageFilterNoDollar = getMessageFilterNoDollar(baseFilter);
+
routingHandler = new MQTTRoutingHandler(server);
sessionStateManager = MQTTStateManager.getInstance(server);
+ sessionStateManager.init();
server.registerActivateCallback(new CleaningActivateCallback() {
@Override
public void deActivate() {
@@ -95,6 +106,30 @@ public class MQTTProtocolManager extends AbstractProtocolManager<MqttMessage, MQ
});
}
+ static StringBuilder getBaseFilter(ActiveMQServer server) {
+ StringBuilder baseFilter = new StringBuilder();
+ baseFilter.append("NOT (");
+ baseFilter.append("(").append(FilterConstants.ACTIVEMQ_ADDRESS).append(" = '").append(server.getConfiguration().getManagementAddress()).append("')");
+ baseFilter.append(" OR ");
+ baseFilter.append("(").append(FilterConstants.ACTIVEMQ_ADDRESS).append(" = '").append(server.getConfiguration().getManagementNotificationAddress()).append("')");
+ return baseFilter;
+ }
+
+ static SimpleString getMessageFilter(StringBuilder baseFilter) {
+ StringBuilder messageFilter = new StringBuilder(baseFilter);
+ messageFilter.append(")");
+ return SimpleString.of(messageFilter.toString());
+ }
+
+ static SimpleString getMessageFilterNoDollar(StringBuilder baseFilter) {
+ // [MQTT-4.7.2-1]
+ StringBuilder messageFilterNoDollar = new StringBuilder(baseFilter);
+ messageFilterNoDollar.append(" OR ");
+ messageFilterNoDollar.append("(").append(FilterConstants.ACTIVEMQ_ADDRESS).append(" LIKE '").append(MQTTUtil.DOLLAR).append("%')");
+ messageFilterNoDollar.append(")");
+ return SimpleString.of(messageFilterNoDollar.toString());
+ }
+
public int getDefaultMqttSessionExpiryInterval() {
return defaultMqttSessionExpiryInterval;
}
@@ -122,6 +157,14 @@ public class MQTTProtocolManager extends AbstractProtocolManager<MqttMessage, MQ
return this;
}
+ public SimpleString getMessageFilter() {
+ return messageFilter;
+ }
+
+ public SimpleString getMessageFilterNoDollar() {
+ return messageFilterNoDollar;
+ }
+
public int getMaximumPacketSize() {
return maximumPacketSize;
}
@@ -216,23 +259,18 @@ public class MQTTProtocolManager extends AbstractProtocolManager<MqttMessage, MQ
@Override
public ConnectionEntry createConnectionEntry(Acceptor acceptorUsed, Connection connection) {
- try {
- MQTTConnection mqttConnection = new MQTTConnection(connection);
- /*
- * We must adjust the keep-alive because MQTT communicates keep-alive values in *seconds*, but the broker uses
- * *milliseconds*. Also, the connection keep-alive is effectively "one and a half times" the configured
- * keep-alive value. See [MQTT-3.1.2-22].
- */
- ConnectionEntry entry = new ConnectionEntry(mqttConnection, null, System.currentTimeMillis(), getServerKeepAlive() == -1 || getServerKeepAlive() == 0 ? -1 : getServerKeepAlive() * MQTTUtil.KEEP_ALIVE_ADJUSTMENT);
-
- NettyServerConnection nettyConnection = ((NettyServerConnection) connection);
- MQTTProtocolHandler protocolHandler = nettyConnection.getChannel().pipeline().get(MQTTProtocolHandler.class);
- protocolHandler.setConnection(mqttConnection, entry);
- return entry;
- } catch (Exception e) {
- logger.error("Error creating connection entry", e);
- return null;
- }
+ MQTTConnection mqttConnection = new MQTTConnection(connection);
+ /*
+ * We must adjust the keep-alive because MQTT communicates keep-alive values in *seconds*, but the broker uses
+ * *milliseconds*. Also, the connection keep-alive is effectively "one and a half times" the configured
+ * keep-alive value. See [MQTT-3.1.2-22].
+ */
+ ConnectionEntry entry = new ConnectionEntry(mqttConnection, null, System.currentTimeMillis(), getServerKeepAlive() == -1 || getServerKeepAlive() == 0 ? -1 : getServerKeepAlive() * MQTTUtil.KEEP_ALIVE_ADJUSTMENT);
+
+ NettyServerConnection nettyConnection = ((NettyServerConnection) connection);
+ MQTTProtocolHandler protocolHandler = nettyConnection.getChannel().pipeline().get(MQTTProtocolHandler.class);
+ protocolHandler.setConnection(mqttConnection, entry);
+ return entry;
}
@Override
@@ -336,6 +374,9 @@ public class MQTTProtocolManager extends AbstractProtocolManager<MqttMessage, MQ
}
public MQTTStateManager getStateManager() {
+ if (sessionStateManager == null) {
+ throw new IllegalStateException("Broker has been deactivated");
+ }
return sessionStateManager;
}
}
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManagerFactory.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManagerFactory.java
index 9d65b1de8d..62e8f2ade0 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManagerFactory.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManagerFactory.java
@@ -66,7 +66,8 @@ public class MQTTProtocolManagerFactory extends AbstractProtocolManagerFactory<M
}
@Override
- public void loadProtocolServices(ActiveMQServer server, List<ActiveMQComponent> services) {
+ public void loadProtocolServices(ActiveMQServer server, List<ActiveMQComponent> services) throws Exception {
+ server.registerRecordsLoader(MQTTStateManager.getInstance(server)::reload);
services.add(new MQTTPeriodicTasks(server, server.getScheduledPool()));
}
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java
index 5f8d8453f8..8c337426f2 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java
@@ -33,17 +33,12 @@ import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttTopicSubscription;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
-import org.apache.activemq.artemis.api.core.ActiveMQIllegalStateException;
import org.apache.activemq.artemis.api.core.ActiveMQSecurityException;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.Message;
-import org.apache.activemq.artemis.api.core.Pair;
-import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
-import org.apache.activemq.artemis.core.io.IOCallback;
import org.apache.activemq.artemis.core.protocol.mqtt.exceptions.DisconnectException;
-import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.ServerConsumer;
import org.apache.activemq.artemis.core.server.ServerProducer;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
@@ -68,7 +63,6 @@ import static org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil.MQTT_PAYLO
import static org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil.MQTT_RESPONSE_TOPIC_KEY;
import static org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil.MQTT_USER_PROPERTY_EXISTS_KEY;
import static org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil.MQTT_USER_PROPERTY_KEY_PREFIX_SIMPLE;
-import static org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil.createServerMessage;
/**
* Handles MQTT Exactly Once (QoS level 2) Protocol.
@@ -77,24 +71,14 @@ public class MQTTPublishManager {
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private SimpleString qos2ManagementAddress;
-
- private Queue qos2ManagementQueue;
-
private final String senderName = UUIDGenerator.getInstance().generateUUID().toString();
private boolean createProducer = true;
- private ServerConsumer qos2ManagementConsumer;
-
private final MQTTSession session;
private final Object lock = new Object();
- private MQTTSessionState state;
-
- private MQTTSessionState.OutboundStore outboundStore;
-
private boolean closeMqttConnectionOnPublishAuthorizationFailure;
public MQTTPublishManager(MQTTSession session, boolean closeMqttConnectionOnPublishAuthorizationFailure) {
@@ -102,58 +86,56 @@ public class MQTTPublishManager {
this.closeMqttConnectionOnPublishAuthorizationFailure = closeMqttConnectionOnPublishAuthorizationFailure;
}
- synchronized void start() {
- this.state = session.getState();
- this.outboundStore = state.getOutboundStore();
- }
-
- synchronized void stop() throws Exception {
+ void stop() throws Exception {
ServerSessionImpl serversession = session.getServerSession();
if (serversession != null) {
serversession.removeProducer(serversession.getName());
}
- if (qos2ManagementConsumer != null) {
- qos2ManagementConsumer.removeItself();
- qos2ManagementConsumer.setStarted(false);
- qos2ManagementConsumer.close(false);
- }
- }
-
- void clean() throws Exception {
- if (qos2ManagementQueue != null) {
- qos2ManagementQueue.deleteQueue();
- }
- }
-
- boolean isQos2ManagementConsumer(ServerConsumer consumer) {
- return consumer == qos2ManagementConsumer;
}
/**
- * Since MQTT Subscriptions can overlap, a client may receive the same message twice. When this happens the client
- * returns a PubRec or PubAck with ID. But we need to know which consumer to ack, since we only have the ID to go on
- * we are not able to decide which consumer to ack. Instead we send MQTT messages with different IDs and store a
- * reference to original ID and consumer in the Session state. This way we can look up the consumer Id and the
- * message Id from the PubAck or PubRec message id.
+ * Delivers a message to the MQTT client at the appropriate QoS level. For QoS 1 and 2, each delivery is tracked by a
+ * journal-persisted {@link PacketIdCorrelationKey} that pairs the core message ID and subscription address to an
+ * MQTT packet ID. This ensures the same packet ID is reused when a message is redelivered after a broker restart, as
+ * required by the MQTT specification. The address component of the key allows overlapping subscriptions to receive
+ * the same message with distinct packet IDs.
+ * <p>
+ * A {@link CoreDeliveryInfo} is also stored in-memory for each in-flight packet ID, mapping it back to the consumer
+ * ID and correlation key so the correct consumer can be acknowledged when the client responds with PUBACK or
+ * PUBCOMP.
*/
- protected void sendMessage(ICoreMessage message, ServerConsumer consumer, int deliveryCount) throws Exception {
- // This is to allow retries of PubRel.
- if (isQos2ManagementConsumer(consumer)) {
- sendPubRelMessage(message);
- } else {
- int qos = decideQoS(message, consumer);
- if (qos == 0) {
- if (publishToClient((int) message.getMessageID(), message, deliveryCount, qos, consumer.getID())) {
- session.getServerSession().individualAcknowledge(consumer.getID(), message.getMessageID());
+ protected void publishToClient(ICoreMessage message, ServerConsumer consumer) throws Exception {
+ MQTTSessionState state = session.getState();
+ int qos = decideQoS(message, consumer);
+ if (qos == 0) {
+ // [MQTT-2.2.1-2] Hard-code the packet ID to 0 as QoS0 PUBLISH packets don't have a packet ID
+ // [MQTT-3.3.1-2] The DUP flag MUST be set to 0 for all QoS 0 messages.
+ if (publishToClient(0, message, false, qos)) {
+ consumer.individualAcknowledge(null, message.getMessageID());
+ }
+ } else if (qos == 1 || qos == 2) {
+ Integer existingPacketId;
+ final int packetIdToUse;
+ boolean redelivery = false;
+ synchronized (this) {
+ PacketIdCorrelationKey correlationKey = PacketIdCorrelationKey.of(message.getMessageID(), message.getAddressSimpleString());
+ existingPacketId = session.getStateManager().getPacketIdCorrelation(state.getClientId(), correlationKey);
+ if (existingPacketId != null && !state.coreDeliveryInfoExists(existingPacketId)) {
+ // re-delivery after reconnect or restart; reuse persisted packet ID
+ packetIdToUse = existingPacketId;
+ redelivery = true;
+ } else {
+ // first delivery, or same message via a different subscription
+ packetIdToUse = state.generatePacketId();
+ session.getStateManager().putPacketIdCorrelation(state.getClientId(), correlationKey, packetIdToUse);
}
- } else if (qos == 1 || qos == 2) {
- int mqttid = outboundStore.generateMqttId(message.getMessageID(), consumer.getID());
- outboundStore.publish(mqttid, message.getMessageID(), consumer.getID());
- publishToClient(mqttid, message, deliveryCount, qos, consumer.getID());
- } else {
- // Client must have disconnected and it's Subscription QoS cleared
- consumer.individualCancel(message.getMessageID(), false);
+ state.putCoreDeliveryInfo(packetIdToUse, CoreDeliveryInfo.of(consumer.getID(), correlationKey));
+ state.incrementSendQuota();
}
+ publishToClient(packetIdToUse, message, redelivery, qos);
+ } else {
+ // Client must have disconnected and it's Subscription QoS cleared
+ consumer.individualCancel(message.getMessageID(), false);
}
}
@@ -205,77 +187,81 @@ public class MQTTPublishManager {
if (qos > 0) {
serverMessage.setDurable(MQTTUtil.DURABLE_MESSAGES);
}
- int packetId = message.variableHeader().packetId();
- boolean qos2PublishAlreadyReceived = state.getPubRec().contains(packetId);
- if (qos < 2 || !qos2PublishAlreadyReceived) {
- Transaction tx = session.getServerSession().newTransaction();
- try {
- AddressInfo addressInfo = session.getServer().getAddressInfo(address);
- if (addressInfo == null && session.getServer().getAddressSettingsRepository().getMatch(coreAddress).isAutoCreateAddresses()) {
- session.getServerSession().createAddress(address, RoutingType.MULTICAST, true);
- serverMessage.setRoutingType(RoutingType.MULTICAST);
- }
- if (addressInfo != null) {
- serverMessage.setRoutingType(addressInfo.getRoutingType());
- }
- session.getServerSession().send(tx, serverMessage, true, senderName, false);
- if (qos == 2 && !internal) {
- state.getPubRec().add(packetId);
- }
+ // only start a transction if really necessary
+ Transaction tx = (qos == 2 && !internal) || message.fixedHeader().isRetain() ? session.getServerSession().newTransaction() : null;
- if (message.fixedHeader().isRetain()) {
- ByteBuf payload = message.payload();
- boolean reset = payload instanceof EmptyByteBuf || payload.capacity() == 0;
- session.getRetainMessageManager().handleRetainedMessage(serverMessage, topic, reset, tx);
- }
+ try {
+ AddressInfo addressInfo = session.getServer().getAddressInfo(address);
+ if (addressInfo == null && session.getServer().getAddressSettingsRepository().getMatch(coreAddress).isAutoCreateAddresses()) {
+ session.getServerSession().createAddress(address, RoutingType.MULTICAST, true);
+ serverMessage.setRoutingType(RoutingType.MULTICAST);
+ }
+ if (addressInfo != null) {
+ serverMessage.setRoutingType(addressInfo.getRoutingType());
+ }
+
+ session.getServerSession().send(tx, serverMessage, true, senderName, false);
+
+ if (qos == 2 && !internal) {
+ session.getState().getPublishCache().add(message.variableHeader().packetId(), tx);
+ }
+
+ if (message.fixedHeader().isRetain()) {
+ ByteBuf payload = message.payload();
+ boolean reset = payload instanceof EmptyByteBuf || payload.capacity() == 0;
+ session.getRetainMessageManager().handleRetainedMessage(serverMessage, topic, reset, tx);
+ }
+ if (tx != null) {
tx.commit();
- } catch (ActiveMQSecurityException e) {
+ }
+ } catch (ActiveMQSecurityException e) {
+ if (tx != null) {
tx.rollback();
- if (internal) {
- throw e;
- }
- if (session.getVersion() == MQTTVersion.MQTT_5) {
- sendMessageAck(internal, qos, packetId, MQTTReasonCodes.NOT_AUTHORIZED);
- return;
- } else if (session.getVersion() == MQTTVersion.MQTT_3_1_1) {
- /*
- * For MQTT 3.1.1 clients:
- *
- * [MQTT-3.3.5-2] If a Server implementation does not authorize a PUBLISH to be performed by a Client;
- * it has no way of informing that Client. It MUST either make a positive acknowledgement, according
- * to the normal QoS rules, or close the Network Connection
- *
- * Throwing an exception here will ultimately close the connection. This is the default behavior.
- */
- if (closeMqttConnectionOnPublishAuthorizationFailure) {
- throw new DisconnectException();
- } else {
- logger.debug("MQTT 3.1.1 client not authorized to publish message.");
- }
+ }
+ if (internal) {
+ throw e;
+ }
+ if (session.getVersion() == MQTTVersion.MQTT_5) {
+ sendMessageAck(internal, qos, message.variableHeader().packetId(), MQTTReasonCodes.NOT_AUTHORIZED);
+ return;
+ } else if (session.getVersion() == MQTTVersion.MQTT_3_1_1) {
+ /*
+ * For MQTT 3.1.1 clients:
+ *
+ * [MQTT-3.3.5-2] If a Server implementation does not authorize a PUBLISH to be performed by a Client;
+ * it has no way of informing that Client. It MUST either make a positive acknowledgement, according
+ * to the normal QoS rules, or close the Network Connection
+ *
+ * Throwing an exception here will ultimately close the connection. This is the default behavior.
+ */
+ if (closeMqttConnectionOnPublishAuthorizationFailure) {
+ throw new DisconnectException();
} else {
- /*
- * For MQTT 3.1 clients:
- *
- * Note that if a server implementation does not authorize a PUBLISH to be made by a client, it has no
- * way of informing that client. It must therefore make a positive acknowledgement, according to the
- * normal QoS rules, and the client will *not* be informed that it was not authorized to publish the
- * message.
- *
- * Log the failure since we have to just swallow it.
- */
- logger.debug("MQTT 3.1 client not authorized to publish message.");
+ logger.debug("MQTT 3.1.1 client not authorized to publish message.");
}
- } catch (Throwable t) {
- MQTTLogger.LOGGER.failedToPublishMqttMessage(t.getMessage(), t);
+ } else {
+ /*
+ * For MQTT 3.1 clients:
+ *
+ * Note that if a server implementation does not authorize a PUBLISH to be made by a client, it has no
+ * way of informing that client. It must therefore make a positive acknowledgement, according to the
+ * normal QoS rules, and the client will *not* be informed that it was not authorized to publish the
+ * message.
+ *
+ * Log the failure since we have to just swallow it.
+ */
+ logger.debug("MQTT 3.1 client not authorized to publish message.");
+ }
+ } catch (Throwable t) {
+ MQTTLogger.LOGGER.failedToPublishMqttMessage(session.getState().getClientId(), message.variableHeader().packetId(), t.getMessage(), t);
+ if (tx != null) {
tx.rollback();
- throw t;
}
- } else if (qos2PublishAlreadyReceived) {
- MQTTLogger.LOGGER.ignoringQoS2Publish(state.getClientId(), packetId);
+ throw t;
}
- createMessageAck(packetId, qos, internal);
+ session.getProtocolHandler().runAfterStorageOperations(() -> sendMessageAck(internal, qos, message.variableHeader().packetId(), MQTTReasonCodes.SUCCESS));
}
}
@@ -289,54 +275,19 @@ public class MQTTPublishManager {
}
}
- void sendPubRelMessage(Message message) {
- int messageId = message.getIntProperty(MQTTUtil.MQTT_MESSAGE_ID_KEY);
- session.getState().getOutboundStore().publishReleasedSent(messageId, message.getMessageID());
- session.getProtocolHandler().sendPubRel(messageId);
- }
-
- void handlePubRec(int messageId) throws Exception {
- try {
- Pair<Long, Long> ref = outboundStore.publishReceived(messageId);
- if (ref != null) {
- initQos2Resources();
- MQTTUtil.sendMessageDirectlyToQueue(session.getServer().getStorageManager(), session.getServer().getPostOffice(), createPubRelMessage(session, messageId), qos2ManagementQueue, null);
- session.getServerSession().individualAcknowledge(ref.getB(), ref.getA());
- releaseFlowControl(ref.getB());
- } else {
- session.getProtocolHandler().sendPubRel(messageId);
- }
- } catch (ActiveMQIllegalStateException e) {
- MQTTLogger.LOGGER.failedToAckMessage(session.getState().getClientId(), e);
- }
+ synchronized void handlePubRecError(int packetId) throws Exception {
+ acknowledgeDelivery(packetId, false);
}
- /*
- * Only create these resources if we actually need them (i.e. we're sending a message to a subscriber via QoS 2)
- */
- private void initQos2Resources() throws Exception {
- if (qos2ManagementAddress == null) {
- qos2ManagementAddress = SimpleString.of(MQTTUtil.QOS2_MANAGEMENT_QUEUE_PREFIX + session.getState().getClientId());
- }
- if (qos2ManagementQueue == null) {
- qos2ManagementQueue = session.getServer().createQueue(QueueConfiguration.of(qos2ManagementAddress)
- .setRoutingType(RoutingType.ANYCAST)
- .setDurable(MQTTUtil.DURABLE_MESSAGES),
- true);
- qos2ManagementConsumer = session.getServerSession().createInternalConsumer(qos2ManagementAddress);
- qos2ManagementConsumer.setStarted(true);
+ synchronized void handlePubRec(int packetId) throws Exception {
+ MQTTSessionState state = session.getState();
+ if (state.getPubRecCache().contains(packetId)) {
+ sendAcknowledgementReply(packetId, MQTTReasonCodes.SUCCESS, true);
+ } else {
+ acknowledgeDelivery(packetId, true);
}
}
- private Message createPubRelMessage(MQTTSession session, int messageId) {
- MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_LEAST_ONCE, false, 0);
- MqttPublishMessage publishMessage = new MqttPublishMessage(fixedHeader, null, null);
- Message message = createServerMessage(session, qos2ManagementAddress, publishMessage)
- .putIntProperty(MQTTUtil.MQTT_MESSAGE_ID_KEY, messageId)
- .putIntProperty(MQTTUtil.MQTT_MESSAGE_TYPE_KEY, MqttMessageType.PUBREL.value());
- return message;
- }
-
/**
* Once we get an acknowledgement for a QoS 1 or 2 message we allow messages to flow
*/
@@ -347,76 +298,105 @@ public class MQTTPublishManager {
}
}
- void handlePubComp(int messageId) throws Exception {
- Pair<Long, Long> ref = session.getState().getOutboundStore().publishComplete(messageId);
- if (ref != null) {
- // ack the message via the internal server session to bypass security.
- session.getServerSession().individualAcknowledge(qos2ManagementConsumer.getID(), ref.getA());
- }
+ synchronized void handlePubComp(int packetId) throws Exception {
+ session.getState().getPubRecCache().remove(packetId);
}
- private void createMessageAck(final int messageId, final int qos, final boolean internal) {
- session.getServer().getStorageManager().afterCompleteOperations(new IOCallback() {
- @Override
- public void done() {
- sendMessageAck(internal, qos, messageId, MQTTReasonCodes.SUCCESS);
- }
-
- @Override
- public void onError(int errorCode, String errorMessage) {
- logger.error("Pub Sync Failed");
- }
- });
+ synchronized void handlePubRel(int packetId) throws Exception {
+ boolean deleted = session.getState().getPublishCache().remove(packetId);
+ if (!deleted) {
+ logger.debug("MQTT client {} sent PUBREL for packet {} but no corresponding PUBLISH was found in the cache", session.getState().getClientId(), packetId);
+ }
+ session.getProtocolHandler().sendPubComp(packetId);
}
- void handlePubRel(int messageId) {
- // We don't check to see if a PubRel existed for this message. We assume it did and so send PubComp.
- state.getPubRec().remove(messageId);
- session.getProtocolHandler().sendPubComp(messageId);
- state.removeMessageRef(messageId);
+ synchronized void handlePubAck(int packetId) throws Exception {
+ acknowledgeDelivery(packetId, false);
}
- void handlePubAck(int messageId) throws Exception {
+ /**
+ * Acknowledges an outbound QoS 1 or QoS 2 delivery when the client confirms receipt.
+ * <p>
+ * For <b>QoS 1</b> no reply is sent since the flow is complete.
+ * <p>
+ * For <b>QoS 2</b> the packet ID is recorded in the PUBREC cache and a PUBREL is sent to continue the handshake. It
+ * is also called when the client's PUBREC carries an error reason code, in which case the delivery is simply cleaned
+ * up with no reply.
+ * <p>
+ * The Core message acknowledgement, packet ID correlation removal, and (for QoS 2) PUBREC cache insertion are
+ * performed atomically in a single transaction.
+ *
+ * @param packetId the MQTT packet identifier for the in-flight delivery
+ * @param needsPubRel {@code true} to record the PUBREC and reply with PUBREL (QoS 2 normal path); {@code false} to
+ * acknowledge silently (QoS 1 or QoS 2 error path)
+ */
+ private void acknowledgeDelivery(int packetId, boolean needsPubRel) throws Exception {
+ MQTTSessionState state = session.getState();
+ Transaction tx = null;
try {
- Pair<Long, Long> ref = outboundStore.publishAckd(messageId);
- if (ref != null) {
- session.getServerSession().individualAcknowledge(ref.getB(), ref.getA());
- releaseFlowControl(ref.getB());
+ CoreDeliveryInfo delivery = state.getCoreDeliveryInfo(packetId);
+ if (delivery != null) {
+ ServerConsumer consumer = session.getServerSession().locateConsumer(delivery.getConsumerId());
+ if (consumer == null) {
+ MQTTLogger.LOGGER.failedToAckMessageConsumerNotFound(state.getClientId(), packetId, delivery.getConsumerId(), session.getServerSession().isClosed() ? "closed" : "not closed");
+ sendAcknowledgementReply(packetId, MQTTReasonCodes.PACKET_IDENTIFIER_NOT_FOUND, needsPubRel);
+ return;
+ }
+ tx = session.getServerSession().newTransaction();
+ if (needsPubRel) {
+ state.getPubRecCache().add(packetId, tx);
+ }
+ session.getStateManager().removePacketIdCorrelation(state.getClientId(), delivery.getPacketIdCorrelationKey(), tx.getID());
+ consumer.individualAcknowledge(tx, delivery.getCoreMessageId());
+ tx.commit();
+ state.removeCoreDeliveryInfo(packetId);
+ state.decrementSendQuota();
+ releaseFlowControl(delivery.getConsumerId());
+ sendAcknowledgementReply(packetId, MQTTReasonCodes.SUCCESS, needsPubRel);
+ } else {
+ sendAcknowledgementReply(packetId, MQTTReasonCodes.PACKET_IDENTIFIER_NOT_FOUND, needsPubRel);
}
- } catch (ActiveMQIllegalStateException e) {
- logger.warn("MQTT Client({}) attempted to Ack already Ack'd message", session.getState().getClientId());
+ } catch (Exception e) {
+ if (tx != null) {
+ tx.rollback();
+ }
+ MQTTLogger.LOGGER.failedToAckMessage(state.getClientId(), e.getMessage());
+ sendAcknowledgementReply(packetId, MQTTReasonCodes.PACKET_IDENTIFIER_NOT_FOUND, needsPubRel);
+ }
+ }
+
+ private void sendAcknowledgementReply(int packetId, byte reasonCode, boolean needsPubRel) throws Exception {
+ if (needsPubRel) {
+ session.getProtocolHandler().sendPubRel(packetId, reasonCode);
}
}
- private boolean publishToClient(int messageId, ICoreMessage message, int deliveryCount, int qos, long consumerId) throws Exception {
- String topic = MQTTUtil.getMqttTopicFromCoreAddress(Objects.requireNonNullElse(message.getAddress(), ""), session.getWildcardConfiguration());
+ private boolean publishToClient(int packetId, ICoreMessage coreMessage, boolean redelivery, int qos) throws Exception {
+ String topic = MQTTUtil.getMqttTopicFromCoreAddress(Objects.requireNonNullElse(coreMessage.getAddress(), ""), session.getWildcardConfiguration());
ByteBuf payload;
- switch (message.getType()) {
+ switch (coreMessage.getType()) {
case Message.TEXT_TYPE:
- SimpleString text = message.getDataBuffer().readNullableSimpleString();
+ SimpleString text = coreMessage.getDataBuffer().readNullableSimpleString();
final int utf8Bytes = ByteBufUtil.utf8Bytes(text);
payload = ByteBufAllocator.DEFAULT.directBuffer(utf8Bytes);
// IMPORTANT: this one won't enlarge ByteBuf by ByteBufUtil.maxUtf8Bytes(text), but just utf8Bytes
ByteBufUtil.reserveAndWriteUtf8(payload, text, utf8Bytes);
break;
default:
- ActiveMQBuffer bodyBuffer = message.getDataBuffer();
+ ActiveMQBuffer bodyBuffer = coreMessage.getDataBuffer();
payload = ByteBufAllocator.DEFAULT.directBuffer(bodyBuffer.writerIndex());
payload.writeBytes(bodyBuffer.byteBuf());
break;
}
- // [MQTT-3.3.1-2] The DUP flag MUST be set to 0 for all QoS 0 messages.
- boolean redelivery = qos == 0 ? false : (deliveryCount > 1);
-
- boolean isRetain = message.containsProperty(MQTT_MESSAGE_RETAIN_INITIAL_DISTRIBUTION_KEY);
+ boolean isRetain = coreMessage.containsProperty(MQTT_MESSAGE_RETAIN_INITIAL_DISTRIBUTION_KEY);
MqttProperties mqttProperties = null;
if (session.getVersion() == MQTTVersion.MQTT_5) {
- mqttProperties = getPublishProperties(message);
- if (!isRetain && message.getBooleanProperty(MQTT_MESSAGE_RETAIN_KEY)) {
- MqttTopicSubscription sub = session.getState().getSubscription(topic);
+ mqttProperties = getPublishProperties(coreMessage);
+ if (!isRetain && coreMessage.getBooleanProperty(MQTT_MESSAGE_RETAIN_KEY)) {
+ MqttTopicSubscription sub = session.getState().getSubscriptionItem(topic).getSubscription();
if (sub != null && sub.option().isRetainAsPublished()) {
isRetain = true;
}
@@ -437,8 +417,8 @@ public class MQTTPublishManager {
}
int remainingLength = MQTTUtil.calculateRemainingLength(topic, mqttProperties, payload);
- MqttFixedHeader header = new MqttFixedHeader(MqttMessageType.PUBLISH, redelivery, MqttQoS.valueOf(qos), isRetain, remainingLength);
- MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic, messageId, mqttProperties);
+ MqttFixedHeader header = new MqttFixedHeader(MqttMessageType.PUBLISH, qos == 0 ? false : redelivery, MqttQoS.valueOf(qos), isRetain, remainingLength);
+ MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic, packetId, mqttProperties);
MqttPublishMessage publish = new MqttPublishMessage(header, varHeader, payload);
int maxSize = session.getState().getClientMaxPacketSize();
@@ -449,8 +429,12 @@ public class MQTTPublishManager {
* [MQTT-3.1.2-25] Where a Packet is too large to send, the Server MUST discard it without sending it and then
* behave as if it had completed sending that Application Message
*/
- logger.debug("Not sending message {} to client as its size ({}) exceeds the max ({})", message, size, maxSize);
- session.getServerSession().individualAcknowledge(consumerId, message.getMessageID());
+ logger.debug("Not sending message {} to client as its size ({}) exceeds the max ({})", coreMessage, size, maxSize);
+ if (qos == 0) {
+ return true;
+ } else if (qos == 1 || qos == 2) {
+ acknowledgeDelivery(packetId, false);
+ }
return false;
}
}
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 551ab41626..68726dd94b 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
@@ -16,6 +16,7 @@
*/
package org.apache.activemq.artemis.core.protocol.mqtt;
+import java.lang.invoke.MethodHandles;
import java.util.UUID;
import io.netty.buffer.EmptyByteBuf;
@@ -33,7 +34,6 @@ import org.apache.activemq.artemis.core.server.impl.ServerSessionImpl;
import org.apache.activemq.artemis.spi.core.protocol.SessionCallback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.lang.invoke.MethodHandles;
public class MQTTSession {
@@ -81,7 +81,7 @@ public class MQTTSession {
MQTTConnection connection,
MQTTProtocolManager protocolManager,
WildcardConfiguration wildcardConfiguration,
- OperationContext sessionContext) throws Exception {
+ OperationContext sessionContext) {
this.protocolHandler = protocolHandler;
this.protocolManager = protocolManager;
this.stateManager = protocolManager.getStateManager();
@@ -92,7 +92,7 @@ public class MQTTSession {
mqttConnectionManager = new MQTTConnectionManager(this);
mqttPublishManager = new MQTTPublishManager(this, protocolManager.isCloseMqttConnectionOnPublishAuthorizationFailure());
sessionCallback = new MQTTSessionCallback(this, connection, protocolManager.getDefaultMaximumInFlightPublishMessages());
- subscriptionManager = new MQTTSubscriptionManager(this, stateManager);
+ subscriptionManager = new MQTTSubscriptionManager(this);
retainMessageManager = new MQTTRetainMessageManager(this);
state = MQTTSessionState.DEFAULT;
@@ -105,7 +105,6 @@ public class MQTTSession {
* which is synchronized with MQTTConnectionManager.disconnect
*/
void start() throws Exception {
- mqttPublishManager.start();
subscriptionManager.start();
stopped = false;
}
@@ -130,7 +129,8 @@ public class MQTTSession {
state.setAttached(false);
state.setDisconnectedTime(System.currentTimeMillis());
state.clearTopicAliases();
- state.getOutboundStore().resetSendQuota();
+ state.resetSendQuota();
+ state.clearCoreDeliveryInfo();
if (getVersion() == MQTTVersion.MQTT_5) {
if (state.getClientSessionExpiryInterval() == 0) {
@@ -231,7 +231,6 @@ public class MQTTSession {
void clean(boolean enforceSecurity) throws Exception {
subscriptionManager.clean(enforceSecurity);
- mqttPublishManager.clean();
state.clear();
}
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionCallback.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionCallback.java
index 9b80d5b66f..5cd3497484 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionCallback.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionCallback.java
@@ -28,7 +28,7 @@ public class MQTTSessionCallback implements SessionCallback {
private final MQTTConnection connection;
private final int defaultMaximumInFlightPublishMessages;
- public MQTTSessionCallback(MQTTSession session, MQTTConnection connection, int defaultMaximumInFlightPublishMessages) throws Exception {
+ public MQTTSessionCallback(MQTTSession session, MQTTConnection connection, int defaultMaximumInFlightPublishMessages) {
this.session = session;
this.connection = connection;
this.defaultMaximumInFlightPublishMessages = defaultMaximumInFlightPublishMessages;
@@ -49,9 +49,9 @@ public class MQTTSessionCallback implements SessionCallback {
ServerConsumer consumer,
int deliveryCount) {
try {
- session.getMqttPublishManager().sendMessage(ref.getMessage().toCore(), consumer, deliveryCount);
+ session.getMqttPublishManager().publishToClient(ref.getMessage().toCore(), consumer);
} catch (Exception e) {
- MQTTLogger.LOGGER.unableToSendMessage(ref, e);
+ MQTTLogger.LOGGER.unableToSendMessage(session.getState().getClientId(), ref, e);
}
return 1;
}
@@ -110,7 +110,7 @@ public class MQTTSessionCallback implements SessionCallback {
* Therefore, enforce flow-control based on the number of pending QoS 1 & 2 messages
*/
int maxInFlightPublishMessages = connection.getReceiveMaximum() > 0 ? connection.getReceiveMaximum() : defaultMaximumInFlightPublishMessages;
- if (ref != null && ref.isDurable() == true && maxInFlightPublishMessages > 0 && session.getState().getOutboundStore().getSendQuota() >= maxInFlightPublishMessages) {
+ if (ref != null && ref.isDurable() == true && maxInFlightPublishMessages > 0 && session.getState().getSendQuota() >= maxInFlightPublishMessages) {
return false;
} else {
return true;
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 8397abee16..031373f1b0 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
@@ -24,9 +24,9 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.mqtt.MqttProperties;
@@ -35,9 +35,7 @@ import io.netty.handler.codec.mqtt.MqttSubscriptionOption;
import io.netty.handler.codec.mqtt.MqttTopicSubscription;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.Message;
-import org.apache.activemq.artemis.api.core.Pair;
import org.apache.activemq.artemis.api.core.SimpleString;
-import org.apache.activemq.artemis.core.config.WildcardConfiguration;
import org.apache.activemq.artemis.core.message.impl.CoreMessage;
import org.apache.activemq.artemis.core.postoffice.Address;
import org.apache.activemq.artemis.core.postoffice.impl.AddressImpl;
@@ -54,20 +52,27 @@ public class MQTTSessionState {
private final String clientId;
- private final ConcurrentMap<String, SubscriptionItem> subscriptions = new ConcurrentHashMap<>();
+ private final ConcurrentMap<String, SubscriptionItem> subscriptionItems = new ConcurrentHashMap<>();
- // Used to store Packet ID of Publish QoS1 and QoS2 message. See spec: 4.3.3 QoS 2: Exactly once delivery. Method B.
- private final Map<Integer, MQTTMessageInfo> messageRefStore = new ConcurrentHashMap<>();
-
- private final ConcurrentMap<String, Map<Long, Integer>> addressMessageMap = new ConcurrentHashMap<>();
+ /**
+ * Records packet IDs of inbound QoS2 PUBLISH messages (client → broker). The ID is added when the broker
+ * receives the PUBLISH and removed when the client completes the handshake with PUBREL, preventing duplicate
+ * processing of the same message.
+ */
+ protected PacketIdCache publishCache;
- private final Set<Integer> pubRec = new HashSet<>();
+ /**
+ * Records packet IDs of outbound QoS2 messages (broker → client) that have reached the PUBREC stage. The
+ * ID is added when the broker receives PUBREC and removed when the client sends PUBCOMP, ensuring the broker can
+ * resume the handshake after a reconnect.
+ */
+ protected PacketIdCache pubRecCache;
private boolean attached = false;
private long disconnectedTime = 0;
- private final OutboundStore outboundStore = new OutboundStore();
+ private PacketIdGenerator packetIdGenerator;
private int clientSessionExpiryInterval;
@@ -97,6 +102,11 @@ public class MQTTSessionState {
private Map<String, Integer> serverTopicAliases;
+ private Map<Integer, CoreDeliveryInfo> coreDeliveryInfos;
+
+ private static final AtomicIntegerFieldUpdater<MQTTSessionState> SEND_QUOTA_UPDATER = AtomicIntegerFieldUpdater.newUpdater(MQTTSessionState.class, "sendQuota");
+ private volatile int sendQuota = 0;
+
public MQTTSessionState(String clientId) {
this.clientId = clientId;
}
@@ -137,7 +147,12 @@ public class MQTTSessionState {
MqttSubscriptionOption.RetainedHandlingPolicy retainedHandlingPolicy = MqttSubscriptionOption.RetainedHandlingPolicy.valueOf(buf.readInt());
Integer subscriptionId = buf.readNullableInt();
- subscriptions.put(topicName, new SubscriptionItem(new MqttTopicSubscription(topicName, new MqttSubscriptionOption(qos, nolocal, retainAsPublished, retainedHandlingPolicy)), subscriptionId));
+ subscriptionItems.put(topicName, SubscriptionItem.of(new MqttTopicSubscription(topicName, new MqttSubscriptionOption(qos, nolocal, retainAsPublished, retainedHandlingPolicy)), subscriptionId));
+ }
+
+ if (buf.readable()) {
+ clientSessionExpiryInterval = buf.readInt();
+ disconnectedTime = System.currentTimeMillis();
}
}
@@ -150,11 +165,16 @@ public class MQTTSessionState {
}
public synchronized void clear() throws Exception {
- subscriptions.clear();
- messageRefStore.clear();
- addressMessageMap.clear();
- pubRec.clear();
- outboundStore.clear();
+ subscriptionItems.clear();
+ if (publishCache != null) {
+ publishCache.clear();
+ }
+ if (pubRecCache != null) {
+ pubRecCache.clear();
+ }
+ if (packetIdGenerator != null) {
+ packetIdGenerator.clear();
+ }
disconnectedTime = 0;
if (willMessage != null) {
willMessage.clear();
@@ -170,12 +190,12 @@ public class MQTTSessionState {
clientTopicAliasMaximum = 0;
}
- public OutboundStore getOutboundStore() {
- return outboundStore;
- }
-
- public Set<Integer> getPubRec() {
- return pubRec;
+ public int generatePacketId() {
+ if (packetIdGenerator == null) {
+ packetIdGenerator = new PacketIdGenerator();
+ }
+ int result = packetIdGenerator.generatePacketId();
+ return result;
}
public boolean isAttached() {
@@ -186,60 +206,32 @@ public class MQTTSessionState {
this.attached = attached;
}
- public Collection<MqttTopicSubscription> getSubscriptions() {
- Collection<MqttTopicSubscription> result = new HashSet<>();
- for (SubscriptionItem item : subscriptions.values()) {
- result.add(item.getSubscription());
- }
- return result;
+ public Map<String, SubscriptionItem> getSubscriptionsPlusID() {
+ return new HashMap<>(subscriptionItems);
}
- public Map<String, SubscriptionItem> getSubscriptionsPlusID() {
- return new HashMap<>(subscriptions);
- }
-
- public boolean addSubscription(MqttTopicSubscription subscription, WildcardConfiguration wildcardConfiguration, Integer subscriptionIdentifier) throws Exception {
- // synchronized to prevent race with removeSubscription
- synchronized (subscriptions) {
- addressMessageMap.putIfAbsent(MQTTUtil.getCoreAddressFromMqttTopic(subscription.topicFilter(), wildcardConfiguration), new ConcurrentHashMap<>());
-
- SubscriptionItem existingSubscription = subscriptions.get(subscription.topicFilter());
- if (existingSubscription != null) {
- if (subscription.qualityOfService().value() > existingSubscription.getSubscription().qualityOfService().value()
- || !Objects.equals(subscriptionIdentifier, existingSubscription.getId())) {
- existingSubscription.update(subscription, subscriptionIdentifier);
- return true;
- } else {
- return false;
- }
- } else {
- subscriptions.put(subscription.topicFilter(), new SubscriptionItem(subscription, subscriptionIdentifier));
- return true;
- }
- }
+ public Collection<SubscriptionItem> getSubscriptionItems() {
+ return new HashSet(subscriptionItems.values());
}
- public void removeSubscription(String address) throws Exception {
- // synchronized to prevent race with addSubscription
- synchronized (subscriptions) {
- subscriptions.remove(address);
- addressMessageMap.remove(address);
- }
+ public void addSubscription(SubscriptionItem item) {
+ subscriptionItems.put(item.getSubscription().topicFilter(), item);
}
- public MqttTopicSubscription getSubscription(String address) {
- return subscriptions.get(address) != null ? subscriptions.get(address).getSubscription() : null;
+ public void removeSubscription(String topicFilter) throws Exception {
+ subscriptionItems.remove(topicFilter);
}
- public SubscriptionItem getSubscriptionPlusID(String address) {
- return subscriptions.get(address);
+ public SubscriptionItem getSubscriptionItem(String topicFilter) {
+ return subscriptionItems.get(topicFilter);
}
public List<Integer> getMatchingSubscriptionIdentifiers(String address) {
String topic = MQTTUtil.getMqttTopicFromCoreAddress(address, session.getServer().getConfiguration().getWildcardConfiguration());
+ Address topicToMatch = new AddressImpl(SimpleString.of(topic), MQTTUtil.MQTT_WILDCARD);
List<Integer> result = null;
- for (SubscriptionItem item : subscriptions.values()) {
- Integer matchingId = item.getMatchingId(topic);
+ for (SubscriptionItem item : subscriptionItems.values()) {
+ Integer matchingId = item.getMatchingId(topicToMatch);
if (matchingId != null) {
if (result == null) {
result = new ArrayList<>();
@@ -394,16 +386,6 @@ public class MQTTSessionState {
return serverTopicAliases == null ? null : serverTopicAliases.get(topicName);
}
- void removeMessageRef(Integer mqttId) {
- MQTTMessageInfo info = messageRefStore.remove(mqttId);
- if (info != null) {
- Map<Long, Integer> addressMap = addressMessageMap.get(info.getAddress());
- if (addressMap != null) {
- addressMap.remove(info.getServerMessageId());
- }
- }
- }
-
public void clearTopicAliases() {
if (clientTopicAliases != null) {
clientTopicAliases.clear();
@@ -415,16 +397,44 @@ public class MQTTSessionState {
}
}
+ public CoreDeliveryInfo getCoreDeliveryInfo(Integer packetId) {
+ return coreDeliveryInfos == null ? null : coreDeliveryInfos.get(packetId);
+ }
+
+ public void putCoreDeliveryInfo(Integer packetId, CoreDeliveryInfo coreDeliveryInfo) {
+ if (coreDeliveryInfos == null) {
+ coreDeliveryInfos = new ConcurrentHashMap<>();
+ }
+ coreDeliveryInfos.put(packetId, coreDeliveryInfo);
+ }
+
+ public CoreDeliveryInfo removeCoreDeliveryInfo(Integer packetId) {
+ if (coreDeliveryInfos != null) {
+ return coreDeliveryInfos.remove(packetId);
+ } else {
+ return null;
+ }
+ }
+
+ public boolean coreDeliveryInfoExists(Integer packetId) {
+ return coreDeliveryInfos == null ? false : coreDeliveryInfos.containsKey(packetId);
+ }
+
+ public void clearCoreDeliveryInfo() {
+ if (coreDeliveryInfos != null) {
+ coreDeliveryInfos.clear();
+ }
+ }
+
@Override
public String toString() {
return "MQTTSessionState[session=" + session +
", clientId=" + clientId +
- ", subscriptions=" + subscriptions +
- ", messageRefStore=" + messageRefStore +
- ", addressMessageMap=" + addressMessageMap +
- ", pubRec=" + pubRec +
+ ", subscriptionItems=" + subscriptionItems +
+ ", publishCache=" + publishCache +
+ ", pubRecCache=" + pubRecCache +
", attached=" + attached +
- ", outboundStore=" + outboundStore +
+ ", packetIdGenerator=" + packetIdGenerator +
", disconnectedTime=" + disconnectedTime +
", sessionExpiryInterval=" + clientSessionExpiryInterval +
", isWill=" + isWill +
@@ -438,104 +448,79 @@ public class MQTTSessionState {
"]@" + System.identityHashCode(this);
}
- public static class OutboundStore {
- private final Map<Pair<Long, Long>, Integer> artemisToMqttMessageMap = new HashMap<>();
-
- private final Map<Integer, Pair<Long, Long>> mqttToServerIds = new HashMap<>();
-
- private final Object dataStoreLock = new Object();
-
- private static final int INITIAL_ID = 0;
-
- private int currentId = INITIAL_ID;
-
- // track send quota independently because it's reset when the client disconnects, but other state must remain in tact
- private int sendQuota = 0;
-
- private Pair<Long, Long> generateKey(long messageId, long consumerID) {
- return new Pair<>(messageId, consumerID);
+ public PacketIdCache getPublishCache() {
+ Objects.requireNonNull(session, "session is null");
+ if (publishCache == null) {
+ publishCache = new PacketIdCache(session, PacketIdCache.TYPE.PUBLISH);
}
+ return publishCache;
+ }
- public int generateMqttId(long messageId, long consumerId) {
- synchronized (dataStoreLock) {
- Integer id = artemisToMqttMessageMap.get(generateKey(messageId, consumerId));
- if (id == null) {
- final int start = currentId;
- do {
- // wrap around to the start if we reach the max
- if (++currentId > MQTTUtil.TWO_BYTE_INT_MAX) {
- currentId = INITIAL_ID;
- }
- // check to see if we looped all the way back around to where we started
- if (start == currentId) {
- // this detects an edge case where the same ID is acked & then generated again
- if (currentId != INITIAL_ID && !mqttToServerIds.containsKey(currentId)) {
- break;
- }
- throw MQTTBundle.BUNDLE.unableToGenerateID();
- }
- }
- while (mqttToServerIds.containsKey(currentId) || currentId == INITIAL_ID);
- id = currentId;
- }
- return id;
- }
+ public PacketIdCache getPubRecCache() {
+ Objects.requireNonNull(session, "session is null");
+ if (pubRecCache == null) {
+ pubRecCache = new PacketIdCache(session, PacketIdCache.TYPE.PUBREC);
}
+ return pubRecCache;
+ }
- public void publish(int mqtt, long messageId, long consumerId) {
- synchronized (dataStoreLock) {
- Pair<Long, Long> key = generateKey(messageId, consumerId);
- artemisToMqttMessageMap.put(key, mqtt);
- mqttToServerIds.put(mqtt, key);
- sendQuota++;
- }
- }
+ public int getSendQuota() {
+ return sendQuota;
+ }
- public Pair<Long, Long> publishAckd(int mqtt) {
- synchronized (dataStoreLock) {
- Pair<Long, Long> p = mqttToServerIds.remove(mqtt);
- if (p != null) {
- sendQuota--;
- artemisToMqttMessageMap.remove(p);
- }
- return p;
- }
- }
+ public void incrementSendQuota() {
+ SEND_QUOTA_UPDATER.incrementAndGet(this);
+ }
- public Pair<Long, Long> publishReceived(int mqtt) {
- return publishAckd(mqtt);
- }
+ public void decrementSendQuota() {
+ SEND_QUOTA_UPDATER.decrementAndGet(this);
+ }
- public void publishReleasedSent(int mqttId, long serverMessageId) {
- synchronized (dataStoreLock) {
- mqttToServerIds.put(mqttId, new Pair<>(serverMessageId, 0L));
- sendQuota++;
- }
- }
+ public void resetSendQuota() {
+ SEND_QUOTA_UPDATER.set(this, 0);
+ }
- public Pair<Long, Long> publishComplete(int mqtt) {
- return publishAckd(mqtt);
- }
+ private class PacketIdGenerator {
+ private static final int INITIAL_ID = 0;
- public void clear() {
- synchronized (dataStoreLock) {
- artemisToMqttMessageMap.clear();
- mqttToServerIds.clear();
- currentId = INITIAL_ID;
- sendQuota = 0;
+ private int currentId = INITIAL_ID;
+
+ private int generatePacketId() {
+ final int start = currentId;
+ do {
+ // wrap around to the start if we reach the max
+ if (++currentId > MQTTUtil.TWO_BYTE_INT_MAX) {
+ currentId = INITIAL_ID;
+ }
+ // check to see if we looped all the way back around to where we started
+ if (start == currentId) {
+ // this detects an edge case where the same ID is acked & then generated again
+ if (currentId != INITIAL_ID && !packetIdInUse(currentId)) {
+ break;
+ }
+ throw MQTTBundle.BUNDLE.unableToGenerateID();
+ }
}
+ while (packetIdInUse(currentId) || currentId == INITIAL_ID);
+ return currentId;
}
- public int getSendQuota() {
- synchronized (dataStoreLock) {
- return sendQuota;
- }
+ /**
+ * Checks to see if the packet ID is in use already for either QoS 1 or QoS 2
+ */
+ private boolean packetIdInUse(int packetId) {
+ // coreDeliveryInfoExists is redundant but is an O(1) short-circuit for the O(n) containsValue in packetIdCorrelationExists
+ return coreDeliveryInfoExists(packetId) ||
+ (session != null &&
+ session.getStateManager() != null &&
+ session.getStateManager().packetIdCorrelationExistsForClient(clientId) &&
+ session.getStateManager().packetIdCorrelationExists(clientId, packetId)) ||
+ (pubRecCache != null &&
+ pubRecCache.contains(packetId));
}
- public void resetSendQuota() {
- synchronized (dataStoreLock) {
- sendQuota = 0;
- }
+ private void clear() {
+ currentId = INITIAL_ID;
}
}
@@ -560,44 +545,4 @@ public class MQTTSessionState {
};
}
}
-
- public static class SubscriptionItem {
- private MqttTopicSubscription subscription;
- private Integer id;
- private Address address;
-
- public SubscriptionItem(MqttTopicSubscription subscription, Integer id) {
- update(subscription, id);
- }
-
- public MqttTopicSubscription getSubscription() {
- return subscription;
- }
-
- public Integer getId() {
- return id;
- }
-
- public Integer getMatchingId(String topic) {
- if (id != null && new AddressImpl(SimpleString.of(topic), MQTTUtil.MQTT_WILDCARD).matches(address)) {
- return id;
- } else {
- return null;
- }
- }
-
- private void update(MqttTopicSubscription newSub, Integer newId) {
- if (newId != null && !newId.equals(id)) {
- if (this.address == null || !subscription.topicFilter().equals(newSub.topicFilter())) {
- String topicFilter = newSub.topicFilter();
- if (MQTTUtil.isSharedSubscription(topicFilter)) {
- topicFilter = MQTTUtil.decomposeSharedSubscriptionTopicFilter(newSub.topicFilter()).getB();
- }
- address = new AddressImpl(SimpleString.of(topicFilter), MQTTUtil.MQTT_WILDCARD);
- }
- }
- subscription = newSub;
- id = newId;
- }
- }
}
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTStateManager.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTStateManager.java
index 99e3d2265f..fcd3d6b171 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTStateManager.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTStateManager.java
@@ -32,8 +32,12 @@ import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.core.filter.impl.FilterImpl;
+import org.apache.activemq.artemis.core.journal.RecordInfo;
+import org.apache.activemq.artemis.core.journal.collections.JournalHashMapProvider;
import org.apache.activemq.artemis.core.message.impl.CoreMessage;
import org.apache.activemq.artemis.core.persistence.StorageManager;
+import org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds;
+import org.apache.activemq.artemis.core.persistence.impl.journal.OperationContextImpl;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.MessageReference;
import org.apache.activemq.artemis.core.server.Queue;
@@ -44,12 +48,13 @@ import org.slf4j.LoggerFactory;
public class MQTTStateManager {
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ private static final Map<Integer, MQTTStateManager> INSTANCES = new HashMap<>();
private final ActiveMQServer server;
private final Map<String, MQTTSessionState> sessionStates = new ConcurrentHashMap<>();
- private final Queue sessionStore;
- private static final Map<Integer, MQTTStateManager> INSTANCES = new HashMap<>();
+ private Queue sessionStore;
private final Map<String, MQTTConnection> connectedClients = new ConcurrentHashMap<>();
private final boolean subscriptionPersistenceEnabled;
+ private final JournalHashMapProvider<String, PacketIdCorrelationKey, Integer, Object> journalHashMapProvider;
/*
* Even though there may be multiple instances of MQTTProtocolManager (e.g. for MQTT on different ports) we only want
@@ -73,37 +78,7 @@ public class MQTTStateManager {
private MQTTStateManager(ActiveMQServer server) throws Exception {
this.server = server;
this.subscriptionPersistenceEnabled = server.getConfiguration().isMqttSubscriptionPersistenceEnabled();
- if (subscriptionPersistenceEnabled) {
- this.sessionStore = server.createQueue(QueueConfiguration.of(MQTTUtil.MQTT_SESSION_STORE).setRoutingType(RoutingType.ANYCAST).setLastValue(true).setDurable(true).setInternal(true).setAutoCreateAddress(true), true);
-
- // load subscription data from queue
- try (LinkedListIterator<MessageReference> iterator = sessionStore.browserIterator()) {
- while (iterator.hasNext()) {
- Message message = iterator.next().getMessage();
- if (!(message instanceof CoreMessage)) {
- MQTTLogger.LOGGER.sessionStateMessageIncorrectType(message.getClass().getName());
- continue;
- }
- String clientId = message.getStringProperty(Message.HDR_LAST_VALUE_NAME);
- if (clientId == null || clientId.isEmpty()) {
- MQTTLogger.LOGGER.sessionStateMessageBadClientId();
- continue;
- }
- MQTTSessionState sessionState;
- try {
- sessionState = new MQTTSessionState((CoreMessage) message);
- } catch (Exception e) {
- MQTTLogger.LOGGER.errorDeserializingStateMessage(e);
- continue;
- }
- sessionStates.put(clientId, sessionState);
- }
- } catch (NoSuchElementException ignored) {
- // this could happen through paging browsing
- }
- } else {
- this.sessionStore = null;
- }
+ this.journalHashMapProvider = new JournalHashMapProvider<>(server.getStorageManager()::generateID, server.getStorageManager(), PacketIdCorrelationKey.getPersister(), JournalRecordIds.MQTT_PACKET_ID_CORRELATION, OperationContextImpl::getContext, null, server.getIoCriticalErrorListener());
}
public void scanSessions() {
@@ -149,9 +124,6 @@ public class MQTTStateManager {
return null;
}
MQTTSessionState removed = sessionStates.remove(clientId);
- if (removed != null && removed.getSubscriptions().size() > 0) {
- removeDurableSubscriptionState(clientId);
- }
return removed;
}
@@ -171,9 +143,9 @@ public class MQTTStateManager {
return "MQTTSessionStateManager@" + Integer.toHexString(System.identityHashCode(this));
}
- public void storeDurableSubscriptionState(MQTTSessionState state) throws Exception {
+ public void storeDurableState(MQTTSessionState state) throws Exception {
if (subscriptionPersistenceEnabled) {
- logger.debug("Adding durable MQTT subscription record for: {}", state.getClientId());
+ logger.debug("Adding durable MQTT record for: {}", state.getClientId());
StorageManager storageManager = server.getStorageManager();
MQTTUtil.sendMessageDirectlyToQueue(storageManager, server.getPostOffice(), serializeState(state, storageManager.generateID()), sessionStore, null);
}
@@ -184,7 +156,7 @@ public class MQTTStateManager {
message.setAddress(MQTTUtil.MQTT_SESSION_STORE);
message.setDurable(true);
message.putStringProperty(Message.HDR_LAST_VALUE_NAME, state.getClientId());
- Map<String, MQTTSessionState.SubscriptionItem> subscriptions = state.getSubscriptionsPlusID();
+ Map<String, SubscriptionItem> subscriptions = state.getSubscriptionsPlusID();
ActiveMQBuffer buf = message.getBodyBuffer();
/*
@@ -195,7 +167,7 @@ public class MQTTStateManager {
buf.writeInt(subscriptions.size());
logger.debug("Serializing {} subscriptions", subscriptions.size());
- for (MQTTSessionState.SubscriptionItem item : subscriptions.values()) {
+ for (SubscriptionItem item : subscriptions.values()) {
MqttTopicSubscription sub = item.getSubscription();
buf.writeString(sub.topicFilter());
buf.writeInt(sub.option().qos().value());
@@ -205,6 +177,8 @@ public class MQTTStateManager {
buf.writeNullableInt(item.getId());
}
+ buf.writeInt(state.getClientSessionExpiryInterval());
+
return message;
}
@@ -244,4 +218,70 @@ public class MQTTStateManager {
public Map<String, MQTTConnection> getConnectedClients() {
return connectedClients;
}
+
+ public void reload(RecordInfo recordInfo) {
+ if (recordInfo.userRecordType == JournalRecordIds.MQTT_PACKET_ID_CORRELATION) {
+ journalHashMapProvider.reload(recordInfo);
+ }
+ }
+
+ public void putPacketIdCorrelation(String clientId, PacketIdCorrelationKey key, Integer packetId) {
+ journalHashMapProvider.getMap(clientId).put(key, packetId);
+ }
+
+ public Integer getPacketIdCorrelation(String clientId, PacketIdCorrelationKey key) {
+ return journalHashMapProvider.getMap(clientId).get(key);
+ }
+
+ public Integer removePacketIdCorrelation(String clientId, PacketIdCorrelationKey key, long transactionId) {
+ return journalHashMapProvider.getMap(clientId).remove(key, transactionId);
+ }
+
+ public void clearPacketIdCorrelation(String clientId) {
+ journalHashMapProvider.getMap(clientId).clear();
+ }
+
+ public boolean packetIdCorrelationExists(String clientId, int packetId) {
+ return journalHashMapProvider.getMap(clientId).containsValue(packetId);
+ }
+
+ public boolean packetIdCorrelationExistsForClient(String clientId) {
+ return journalHashMapProvider.containsMap(clientId);
+ }
+
+ public int getPacketIdCorrelationSize(String clientId) {
+ return journalHashMapProvider.getMap(clientId).size();
+ }
+
+ public void init() throws Exception {
+ if (subscriptionPersistenceEnabled) {
+ this.sessionStore = server.createQueue(QueueConfiguration.of(MQTTUtil.MQTT_SESSION_STORE).setRoutingType(RoutingType.ANYCAST).setLastValue(true).setDurable(true).setInternal(true).setAutoCreateAddress(true), true);
+
+ // load subscription data from queue
+ try (LinkedListIterator<MessageReference> iterator = sessionStore.browserIterator()) {
+ while (iterator.hasNext()) {
+ Message message = iterator.next().getMessage();
+ if (!(message instanceof CoreMessage)) {
+ MQTTLogger.LOGGER.sessionStateMessageIncorrectType(message.getClass().getName());
+ continue;
+ }
+ String clientId = message.getStringProperty(Message.HDR_LAST_VALUE_NAME);
+ if (clientId == null || clientId.isEmpty()) {
+ MQTTLogger.LOGGER.sessionStateMessageBadClientId();
+ continue;
+ }
+ MQTTSessionState sessionState;
+ try {
+ sessionState = new MQTTSessionState((CoreMessage) message);
+ } catch (Exception e) {
+ MQTTLogger.LOGGER.errorDeserializingStateMessage(e);
+ continue;
+ }
+ sessionStates.put(clientId, sessionState);
+ }
+ } catch (NoSuchElementException ignored) {
+ // this could happen through paging browsing
+ }
+ }
+ }
}
\ No newline at end of file
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSubscriptionManager.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSubscriptionManager.java
index 6c65292a5c..4ddf208d74 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSubscriptionManager.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSubscriptionManager.java
@@ -28,7 +28,6 @@ import io.netty.handler.codec.mqtt.MqttSubscriptionOption;
import io.netty.handler.codec.mqtt.MqttTopicSubscription;
import org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException;
import org.apache.activemq.artemis.api.core.ActiveMQSecurityException;
-import org.apache.activemq.artemis.api.core.FilterConstants;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
@@ -39,7 +38,8 @@ import org.apache.activemq.artemis.core.server.ServerConsumer;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.utils.CompositeAddress;
-import static org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil.DOLLAR;
+import static io.netty.handler.codec.mqtt.MqttSubscriptionOption.RetainedHandlingPolicy.SEND_AT_SUBSCRIBE;
+import static io.netty.handler.codec.mqtt.MqttSubscriptionOption.RetainedHandlingPolicy.SEND_AT_SUBSCRIBE_IF_NOT_YET_EXISTS;
import static org.apache.activemq.artemis.reader.MessageUtil.CONNECTION_ID_PROPERTY_NAME_STRING;
public class MQTTSubscriptionManager {
@@ -50,85 +50,74 @@ public class MQTTSubscriptionManager {
private final ConcurrentMap<Long, Integer> consumerQoSLevels;
- private final ConcurrentMap<String, ServerConsumer> consumers;
-
- // We filter out certain messages (e.g. management messages, notifications)
- private final SimpleString messageFilter;
-
- /*
- * We can also filter out messages from any address starting with '$'. This is because MQTT clients can do silly
- * things like subscribe to '#' which matches ever address on the broker.
- */
- private final SimpleString messageFilterNoDollar;
-
private final char singleWord;
private final char anyWords;
- public MQTTSubscriptionManager(MQTTSession session, MQTTStateManager stateManager) {
+ private boolean started = false;
+
+ public MQTTSubscriptionManager(MQTTSession session) {
this.session = session;
- this.stateManager = stateManager;
-
- singleWord = session.getServer().getConfiguration().getWildcardConfiguration().getSingleWord();
- anyWords = session.getServer().getConfiguration().getWildcardConfiguration().getAnyWords();
-
- consumers = new ConcurrentHashMap<>();
- consumerQoSLevels = new ConcurrentHashMap<>();
-
- // Create filter string to ignore certain messages
- StringBuilder baseFilter = new StringBuilder();
- baseFilter.append("NOT (");
- baseFilter.append("(").append(FilterConstants.ACTIVEMQ_ADDRESS).append(" = '").append(session.getServer().getConfiguration().getManagementAddress()).append("')");
- baseFilter.append(" OR ");
- baseFilter.append("(").append(FilterConstants.ACTIVEMQ_ADDRESS).append(" = '").append(session.getServer().getConfiguration().getManagementNotificationAddress()).append("')");
-
- StringBuilder messageFilter = new StringBuilder(baseFilter);
- messageFilter.append(")");
- this.messageFilter = SimpleString.of(messageFilter.toString());
-
- // [MQTT-4.7.2-1]
- StringBuilder messageFilterNoDollar = new StringBuilder(baseFilter);
- messageFilterNoDollar.append(" OR ");
- messageFilterNoDollar.append("(").append(FilterConstants.ACTIVEMQ_ADDRESS).append(" LIKE '").append(DOLLAR).append("%')");
- messageFilterNoDollar.append(")");
- this.messageFilterNoDollar = SimpleString.of(messageFilterNoDollar.toString());
+ this.stateManager = session.getStateManager();
+ this.singleWord = session.getServer().getConfiguration().getWildcardConfiguration().getSingleWord();
+ this.anyWords = session.getServer().getConfiguration().getWildcardConfiguration().getAnyWords();
+ this.consumerQoSLevels = new ConcurrentHashMap<>();
}
- synchronized void start() throws Exception {
- for (MqttTopicSubscription subscription : session.getState().getSubscriptions()) {
- addSubscription(subscription, null, true);
+ void start() throws Exception {
+ MQTTSessionState state = session.getState();
+ synchronized (state) {
+ if (started) {
+ return;
+ }
+
+ for (SubscriptionItem item : state.getSubscriptionItems()) {
+ addSubscription(item.getSubscription(), null, true);
+ }
+
+ // Re-send PUBREL for any QoS 2 messages where PUBREC was received but PUBCOMP wasn't
+ for (int packetId : state.getPubRecCache().getPacketIds()) {
+ session.getProtocolHandler().sendPubRel(packetId, MQTTReasonCodes.SUCCESS);
+ }
+
+ started = true;
}
}
private void addSubscription(MqttTopicSubscription subscription, Integer subscriptionIdentifier, boolean initialStart) throws Exception {
- String rawTopicName = CompositeAddress.extractAddressName(subscription.topicFilter());
- String parsedTopicName = MQTTUtil.decomposeSharedSubscriptionTopicFilter(rawTopicName).getB();
- boolean isFullyQualified = CompositeAddress.isFullyQualified(subscription.topicFilter());
+ final MQTTSessionState state = session.getState();
+ final String topicFilter = subscription.topicFilter();
+ final int qos = subscription.qualityOfService().value();
+ final boolean isNoLocal = subscription.option().isNoLocal();
- Queue q = createQueueForSubscription(rawTopicName, parsedTopicName, isFullyQualified);
+ final String rawTopicName = CompositeAddress.extractAddressName(topicFilter);
+ final String parsedTopicName = MQTTUtil.decomposeSharedSubscriptionTopicFilter(rawTopicName).getB();
- int qos = subscription.qualityOfService().value();
+ Queue q = createQueueForSubscription(rawTopicName, parsedTopicName, CompositeAddress.isFullyQualified(topicFilter));
try {
if (initialStart) {
- createConsumerForSubscriptionQueue(q, parsedTopicName, qos, subscription.option().isNoLocal(), null);
+ // it's safe to assume the subscription exists here because the caller references the same underlying map
+ state.getSubscriptionItem(topicFilter).setConsumer(createConsumer(q, qos, isNoLocal));
} else {
- MqttTopicSubscription existingSubscription = session.getState().getSubscription(parsedTopicName);
- if (existingSubscription == null) {
- createConsumerForSubscriptionQueue(q, parsedTopicName, qos, subscription.option().isNoLocal(), null);
+ SubscriptionItem existingSub = state.getSubscriptionItem(topicFilter);
+
+ if (existingSub == null) {
+ state.addSubscription(SubscriptionItem.of(subscription, subscriptionIdentifier, createConsumer(q, qos, isNoLocal)));
} else {
- Long existingConsumerId = consumers.get(parsedTopicName).getID();
- consumerQoSLevels.put(existingConsumerId, qos);
- if (existingSubscription.option().isNoLocal() != subscription.option().isNoLocal()) {
- createConsumerForSubscriptionQueue(q, parsedTopicName, qos, subscription.option().isNoLocal(), existingConsumerId);
+ ServerConsumer existingConsumer = existingSub.getConsumer();
+ consumerQoSLevels.put(existingConsumer.getID(), qos);
+ if (existingSub.getSubscription().option().isNoLocal() != isNoLocal) {
+ existingSub.setConsumer(createConsumer(q, qos, isNoLocal, existingConsumer.getID()));
+ closeConsumer(existingConsumer);
}
+ existingSub.update(subscription, subscriptionIdentifier);
}
- if (subscription.option().retainHandling() == MqttSubscriptionOption.RetainedHandlingPolicy.SEND_AT_SUBSCRIBE || (subscription.option().retainHandling() == MqttSubscriptionOption.RetainedHandlingPolicy.SEND_AT_SUBSCRIBE_IF_NOT_YET_EXISTS && existingSubscription == null)) {
+ MqttSubscriptionOption.RetainedHandlingPolicy retainedHandlingPolicy = subscription.option().retainHandling();
+ if (retainedHandlingPolicy == SEND_AT_SUBSCRIBE || (retainedHandlingPolicy == SEND_AT_SUBSCRIBE_IF_NOT_YET_EXISTS && existingSub == null)) {
session.getRetainMessageManager().addRetainedMessagesToQueue(q, parsedTopicName);
}
-
- session.getState().addSubscription(subscription, session.getWildcardConfiguration(), subscriptionIdentifier);
}
} catch (Exception e) {
// if anything broke during the creation of the consumer (or otherwise) then ensure the subscription queue
@@ -137,12 +126,20 @@ public class MQTTSubscriptionManager {
}
}
- synchronized void stop() throws Exception {
- for (ServerConsumer consumer : consumers.values()) {
- consumer.setStarted(false);
- consumer.disconnect();
- consumer.getQueue().removeConsumer(consumer);
- consumer.close(false);
+ void stop() throws Exception {
+ MQTTSessionState state = session.getState();
+ synchronized (state) {
+ for (SubscriptionItem item : state.getSubscriptionItems()) {
+ ServerConsumer consumer = item.getConsumer();
+ if (consumer != null && !consumer.isClosed()) {
+ consumer.setStarted(false);
+ consumer.disconnect();
+ consumer.getQueue().removeConsumer(consumer);
+ consumer.close(false);
+ }
+ item.setConsumer(null);
+ }
+ started = false;
}
}
@@ -215,28 +212,38 @@ public class MQTTSubscriptionManager {
* By the time we get here wildcards in the MQTT topic filter have already been translated into their core
* equivalents. This check is to enforce [MQTT-4.7.2-1].
*/
+ MQTTProtocolManager protocolManager = session.getProtocolManager();
if (addressName.startsWith(singleWord) || addressName.startsWith(anyWords)) {
- return messageFilterNoDollar;
+ return protocolManager.getMessageFilterNoDollar();
} else {
- return messageFilter;
+ return protocolManager.getMessageFilter();
}
}
- private void createConsumerForSubscriptionQueue(Queue queue, String topicFilter, int qos, boolean noLocal, Long existingConsumerId) throws Exception {
+ private ServerConsumer createConsumer(Queue queue, int qos, boolean noLocal) throws Exception {
+ return createConsumer(queue, qos, noLocal, null);
+ }
+
+ private ServerConsumer createConsumer(Queue queue, int qos, boolean noLocal, Long existingConsumerId) throws Exception {
long cid = Objects.requireNonNullElseGet(existingConsumerId, () -> session.getServer().getStorageManager().generateID());
// for noLocal support we use the MQTT *client id* rather than the connection ID, but we still use the existing property name
- ServerConsumer consumer = session.getServerSession().createConsumer(cid, queue.getName(), noLocal ? SimpleString.of(CONNECTION_ID_PROPERTY_NAME_STRING + " <> '" + session.getState().getClientId() + "'") : null, false, false, -1);
+ SimpleString filterString = noLocal ? SimpleString.of(CONNECTION_ID_PROPERTY_NAME_STRING + " <> '" + session.getState().getClientId() + "'") : null;
- ServerConsumer existingConsumer = consumers.put(topicFilter, consumer);
- if (existingConsumer != null) {
- existingConsumer.setStarted(false);
- existingConsumer.close(false);
- }
+ ServerConsumer consumer = session.getServerSession().createConsumer(cid, queue.getName(), filterString, false, false, -1);
consumer.setStarted(true);
consumerQoSLevels.put(cid, qos);
+
+ return consumer;
+ }
+
+ private void closeConsumer(ServerConsumer consumer) throws Exception {
+ if (consumer != null) {
+ consumer.setStarted(false);
+ consumer.close(false);
+ }
}
short[] removeSubscriptions(List<String> topics, boolean enforceSecurity) throws Exception {
@@ -250,7 +257,7 @@ public class MQTTSubscriptionManager {
synchronized (state) {
reasonCodes = new short[topics.size()];
for (int i = 0; i < topics.size(); i++) {
- if (state.getSubscription(topics.get(i)) == null) {
+ if (state.getSubscriptionItem(topics.get(i)) == null) {
reasonCodes[i] = MQTTReasonCodes.NO_SUBSCRIPTION_EXISTED;
continue;
}
@@ -258,8 +265,9 @@ public class MQTTSubscriptionManager {
short reasonCode = MQTTReasonCodes.SUCCESS;
try {
+ SubscriptionItem item = state.getSubscriptionItem(topics.get(i));
+ ServerConsumer removed = item != null ? item.getConsumer() : null;
state.removeSubscription(topics.get(i));
- ServerConsumer removed = consumers.remove(MQTTUtil.decomposeSharedSubscriptionTopicFilter(topics.get(i)).getB());
if (removed != null) {
removed.close(false);
consumerQoSLevels.remove(removed.getID());
@@ -283,9 +291,9 @@ public class MQTTSubscriptionManager {
}
// deal with durable state after *all* requested subscriptions have been removed in memory
- if (state.getSubscriptions().size() > 0) {
+ if (state.getSubscriptionItems().size() > 0) {
// if there are some subscriptions left then update the state
- stateManager.storeDurableSubscriptionState(state);
+ stateManager.storeDurableState(state);
} else {
// if there are no subscriptions left then remove the state entirely
stateManager.removeDurableSubscriptionState(state.getClientId());
@@ -303,6 +311,7 @@ public class MQTTSubscriptionManager {
int[] addSubscriptions(List<MqttTopicSubscription> subscriptions, Integer subscriptionIdentifier) throws Exception {
MQTTSessionState state = session.getState();
synchronized (state) {
+ start();
int[] qos = new int[subscriptions.size()];
for (int i = 0; i < subscriptions.size(); i++) {
@@ -338,7 +347,7 @@ public class MQTTSubscriptionManager {
}
// store state after *all* requested subscriptions have been created in memory
- stateManager.storeDurableSubscriptionState(state);
+ stateManager.storeDurableState(state);
return qos;
}
@@ -350,9 +359,10 @@ public class MQTTSubscriptionManager {
void clean(boolean enforceSecurity) throws Exception {
List<String> topics = new ArrayList<>();
- for (MqttTopicSubscription mqttTopicSubscription : session.getState().getSubscriptions()) {
- topics.add(mqttTopicSubscription.topicFilter());
+ for (SubscriptionItem item : session.getState().getSubscriptionItems()) {
+ topics.add(item.getSubscription().topicFilter());
}
removeSubscriptions(topics, enforceSecurity);
+ stateManager.clearPacketIdCorrelation(session.getState().getClientId());
}
}
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTUtil.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTUtil.java
index 3316e55f82..e4c4764f0f 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTUtil.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTUtil.java
@@ -74,9 +74,9 @@ public class MQTTUtil {
public static final boolean DURABLE_MESSAGES = true;
- public static final boolean SESSION_AUTO_COMMIT_SENDS = true;
+ public static final boolean SESSION_AUTO_COMMIT_SENDS = false;
- public static final boolean SESSION_AUTO_COMMIT_ACKS = true;
+ public static final boolean SESSION_AUTO_COMMIT_ACKS = false;
public static final boolean SESSION_PREACKNOWLEDGE = false;
@@ -98,10 +98,6 @@ public class MQTTUtil {
public static final SimpleString MQTT_QOS_LEVEL_KEY = SimpleString.of("mqtt.qos.level");
- public static final SimpleString MQTT_MESSAGE_ID_KEY = SimpleString.of("mqtt.message.id");
-
- public static final SimpleString MQTT_MESSAGE_TYPE_KEY = SimpleString.of("mqtt.message.type");
-
public static final SimpleString MQTT_MESSAGE_RETAIN_KEY = SimpleString.of("mqtt.message.retain");
public static final SimpleString MQTT_MESSAGE_RETAIN_INITIAL_DISTRIBUTION_KEY = SimpleString.of("mqtt.message.retain.initial.distribution");
@@ -120,8 +116,6 @@ public class MQTTUtil {
public static final SimpleString MQTT_CONTENT_TYPE_KEY = SimpleString.of("mqtt.content.type");
- public static final String QOS2_MANAGEMENT_QUEUE_PREFIX = DOLLAR + "sys.mqtt.queue.qos2.";
-
public static final String SHARED_SUBSCRIPTION_PREFIX = DOLLAR + "share/";
public static final long FOUR_BYTE_INT_MAX = Long.decode("0xFFFFFFFF"); // 4_294_967_295
@@ -303,118 +297,130 @@ public class MQTTUtil {
log.append("): OUT >> ");
}
- if (message.fixedHeader() != null) {
- log.append(message.fixedHeader().messageType().toString());
+ String messageForLogging = getMessageForLogging(message, version);
- if (message.variableHeader() instanceof MqttMessageIdVariableHeader) {
- log.append("(" + ((MqttMessageIdVariableHeader) message.variableHeader()).messageId() + ")");
- }
+ if (messageForLogging != null) {
+ logger.trace(log.append(messageForLogging).toString());
+ }
+ }
+ }
- switch (message.fixedHeader().messageType()) {
- case PUBLISH:
- MqttPublishVariableHeader publishHeader = (MqttPublishVariableHeader) message.variableHeader();
- String topicName = publishHeader.topicName();
- if (topicName == null || topicName.isEmpty()) {
- topicName = "<empty>";
- }
- log.append("(" + publishHeader.packetId() + ")")
- .append(" topic=" + topicName)
- .append(", qos=" + message.fixedHeader().qosLevel().value())
- .append(", retain=" + message.fixedHeader().isRetain())
- .append(", dup=" + message.fixedHeader().isDup())
- .append(", remainingLength=" + message.fixedHeader().remainingLength());
- for (MqttProperties.MqttProperty property : ((MqttPublishMessage)message).variableHeader().properties().listAll()) {
- Object value = property.value();
- if (value != null) {
- if (value instanceof byte[] bytes) {
- value = new String(bytes, StandardCharsets.UTF_8);
- } else if (value instanceof ArrayList<?> list && !list.isEmpty() && list.get(0) instanceof MqttProperties.StringPair) {
- StringBuilder userProperties = new StringBuilder();
- userProperties.append("[");
- for (MqttProperties.StringPair pair : (ArrayList<MqttProperties.StringPair>) value) {
- userProperties.append(pair.key).append(": ").append(pair.value).append(", ");
- }
- userProperties.delete(userProperties.length() - 2, userProperties.length());
- userProperties.append("]");
- value = userProperties.toString();
+ public static String getMessageForLogging(MqttMessage message, MQTTVersion version) {
+ String result = null;
+
+ if (message.fixedHeader() != null) {
+ StringBuilder log = new StringBuilder();
+ log.append(message.fixedHeader().messageType().toString());
+
+ if (message.variableHeader() instanceof MqttMessageIdVariableHeader) {
+ log.append("(" + ((MqttMessageIdVariableHeader) message.variableHeader()).messageId() + ")");
+ }
+
+ switch (message.fixedHeader().messageType()) {
+ case PUBLISH:
+ MqttPublishVariableHeader publishHeader = (MqttPublishVariableHeader) message.variableHeader();
+ String topicName = publishHeader.topicName();
+ if (topicName == null || topicName.isEmpty()) {
+ topicName = "<empty>";
+ }
+ log.append("(" + publishHeader.packetId() + ")")
+ .append(" topic=" + topicName)
+ .append(", qos=" + message.fixedHeader().qosLevel().value())
+ .append(", retain=" + message.fixedHeader().isRetain())
+ .append(", dup=" + message.fixedHeader().isDup())
+ .append(", remainingLength=" + message.fixedHeader().remainingLength());
+ for (MqttProperties.MqttProperty property : ((MqttPublishMessage) message).variableHeader().properties().listAll()) {
+ Object value = property.value();
+ if (value != null) {
+ if (value instanceof byte[] bytes) {
+ value = new String(bytes, StandardCharsets.UTF_8);
+ } else if (value instanceof ArrayList<?> list && !list.isEmpty() && list.get(0) instanceof MqttProperties.StringPair) {
+ StringBuilder userProperties = new StringBuilder();
+ userProperties.append("[");
+ for (MqttProperties.StringPair pair : (ArrayList<MqttProperties.StringPair>) value) {
+ userProperties.append(pair.key).append(": ").append(pair.value).append(", ");
}
+ userProperties.delete(userProperties.length() - 2, userProperties.length());
+ userProperties.append("]");
+ value = userProperties.toString();
}
- log.append(", " + formatCase(MqttPropertyType.valueOf(property.propertyId()).name()) + "=" + value);
- }
- log.append(", payload=" + getPayloadForLogging((MqttPublishMessage) message, 256));
- break;
- case CONNECT:
- // intentionally omit the username & password from the log
- MqttConnectVariableHeader connectHeader = (MqttConnectVariableHeader) message.variableHeader();
- MqttConnectPayload payload = ((MqttConnectMessage)message).payload();
- log.append(" protocol=(").append(connectHeader.name()).append(", ").append(connectHeader.version()).append(")")
- .append(", hasPassword=").append(connectHeader.hasPassword())
- .append(", isCleanStart=").append(connectHeader.isCleanSession())
- .append(", keepAliveTimeSeconds=").append(connectHeader.keepAliveTimeSeconds())
- .append(", clientIdentifier=").append(payload.clientIdentifier())
- .append(", hasUserName=").append(connectHeader.hasUserName())
- .append(", isWillFlag=").append(connectHeader.isWillFlag());
- if (connectHeader.isWillFlag()) {
- log.append(", willQos=").append(connectHeader.willQos())
- .append(", isWillRetain=").append(connectHeader.isWillRetain())
- .append(", willTopic=").append(payload.willTopic());
- }
- for (MqttProperties.MqttProperty property : connectHeader.properties().listAll()) {
- log.append(", " + formatCase(MqttPropertyType.valueOf(property.propertyId()).name()) + "=" + property.value());
- }
- break;
- case CONNACK:
- MqttConnAckVariableHeader connackHeader = (MqttConnAckVariableHeader) message.variableHeader();
- log.append(" connectReasonCode=").append(formatByte(connackHeader.connectReturnCode().byteValue()))
- .append(", sessionPresent=").append(connackHeader.isSessionPresent());
- for (MqttProperties.MqttProperty property : connackHeader.properties().listAll()) {
- log.append(", " + formatCase(MqttPropertyType.valueOf(property.propertyId()).name()) + "=" + property.value());
- }
- break;
- case SUBSCRIBE:
- for (MqttTopicSubscription sub : ((MqttSubscribeMessage) message).payload().topicSubscriptions()) {
- log.append("\n\ttopic: ").append(sub.topicName())
- .append(", qos: ").append(sub.qualityOfService())
- .append(", nolocal: ").append(sub.option().isNoLocal())
- .append(", retainHandling: ").append(sub.option().retainHandling())
- .append(", isRetainAsPublished: ").append(sub.option().isRetainAsPublished());
}
- break;
- case SUBACK:
- for (Integer qos : ((MqttSubAckMessage) message).payload().grantedQoSLevels()) {
- log.append("\n\t" + qos);
- }
- break;
- case UNSUBSCRIBE:
- for (String topic : ((MqttUnsubscribeMessage) message).payload().topics()) {
- log.append("\n\t" + topic);
- }
- break;
- case PUBACK:
- break;
- case PUBREC:
- case PUBREL:
- case PUBCOMP:
- if (version == MQTTVersion.MQTT_5) {
- MqttPubReplyMessageVariableHeader pubReplyVariableHeader = (MqttPubReplyMessageVariableHeader) message.variableHeader();
- log.append(" reasonCode=").append(formatByte(pubReplyVariableHeader.reasonCode()));
- }
- break;
- case DISCONNECT:
- if (version == MQTTVersion.MQTT_5) {
- MqttReasonCodeAndPropertiesVariableHeader disconnectVariableHeader = (MqttReasonCodeAndPropertiesVariableHeader) message.variableHeader();
- log.append(" reasonCode=").append(formatByte(disconnectVariableHeader.reasonCode()));
- }
- break;
- }
-
- logger.trace(log.toString());
+ log.append(", " + formatCase(MqttPropertyType.valueOf(property.propertyId()).name()) + "=" + value);
+ }
+ log.append(", payload=" + getPayloadForLogging((MqttPublishMessage) message, 256));
+ break;
+ case CONNECT:
+ // intentionally omit the username & password from the log
+ MqttConnectVariableHeader connectHeader = (MqttConnectVariableHeader) message.variableHeader();
+ MqttConnectPayload payload = ((MqttConnectMessage) message).payload();
+ log.append(" protocol=(").append(connectHeader.name()).append(", ").append(connectHeader.version()).append(")")
+ .append(", hasPassword=").append(connectHeader.hasPassword())
+ .append(", isCleanStart=").append(connectHeader.isCleanSession())
+ .append(", keepAliveTimeSeconds=").append(connectHeader.keepAliveTimeSeconds())
+ .append(", clientIdentifier=").append(payload.clientIdentifier())
+ .append(", hasUserName=").append(connectHeader.hasUserName())
+ .append(", isWillFlag=").append(connectHeader.isWillFlag());
+ if (connectHeader.isWillFlag()) {
+ log.append(", willQos=").append(connectHeader.willQos())
+ .append(", isWillRetain=").append(connectHeader.isWillRetain())
+ .append(", willTopic=").append(payload.willTopic());
+ }
+ for (MqttProperties.MqttProperty property : connectHeader.properties().listAll()) {
+ log.append(", " + formatCase(MqttPropertyType.valueOf(property.propertyId()).name()) + "=" + property.value());
+ }
+ break;
+ case CONNACK:
+ MqttConnAckVariableHeader connackHeader = (MqttConnAckVariableHeader) message.variableHeader();
+ log.append(" connectReasonCode=").append(formatByte(connackHeader.connectReturnCode().byteValue()))
+ .append(", sessionPresent=").append(connackHeader.isSessionPresent());
+ for (MqttProperties.MqttProperty property : connackHeader.properties().listAll()) {
+ log.append(", " + formatCase(MqttPropertyType.valueOf(property.propertyId()).name()) + "=" + property.value());
+ }
+ break;
+ case SUBSCRIBE:
+ for (MqttTopicSubscription sub : ((MqttSubscribeMessage) message).payload().topicSubscriptions()) {
+ log.append("\n\ttopic: ").append(sub.topicName())
+ .append(", qos: ").append(sub.qualityOfService())
+ .append(", nolocal: ").append(sub.option().isNoLocal())
+ .append(", retainHandling: ").append(sub.option().retainHandling())
+ .append(", isRetainAsPublished: ").append(sub.option().isRetainAsPublished());
+ }
+ break;
+ case SUBACK:
+ for (Integer qos : ((MqttSubAckMessage) message).payload().grantedQoSLevels()) {
+ log.append("\n\t" + qos);
+ }
+ break;
+ case UNSUBSCRIBE:
+ for (String topic : ((MqttUnsubscribeMessage) message).payload().topics()) {
+ log.append("\n\t" + topic);
+ }
+ break;
+ case PUBACK:
+ break;
+ case PUBREC:
+ case PUBREL:
+ case PUBCOMP:
+ if (version == MQTTVersion.MQTT_5) {
+ MqttPubReplyMessageVariableHeader pubReplyVariableHeader = (MqttPubReplyMessageVariableHeader) message.variableHeader();
+ log.append(" reasonCode=").append(formatByte(pubReplyVariableHeader.reasonCode()));
+ }
+ break;
+ case DISCONNECT:
+ if (version == MQTTVersion.MQTT_5) {
+ MqttReasonCodeAndPropertiesVariableHeader disconnectVariableHeader = (MqttReasonCodeAndPropertiesVariableHeader) message.variableHeader();
+ log.append(" reasonCode=").append(formatByte(disconnectVariableHeader.reasonCode()));
+ }
+ break;
}
+ result = log.toString();
}
+
+ return result;
}
private static String formatByte(byte bite) {
- return String.format("0x%02X ", bite);
+ return String.format("0x%02X", bite);
}
private static String formatCase(String string) {
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/PacketIdCache.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/PacketIdCache.java
new file mode 100644
index 0000000000..d584a2c82d
--- /dev/null
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/PacketIdCache.java
@@ -0,0 +1,116 @@
+/*
+ * 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.core.protocol.mqtt;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.activemq.artemis.api.core.Pair;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.postoffice.DuplicateIDCache;
+import org.apache.activemq.artemis.core.postoffice.PostOffice;
+import org.apache.activemq.artemis.core.transaction.Transaction;
+import org.apache.activemq.artemis.utils.ByteUtil;
+
+/**
+ * Tracks MQTT packet identifiers using a {@link DuplicateIDCache} to provide durable, duplicate-safe tracking across
+ * broker restarts for QoS2 message flows.
+ * <p>
+ * QoS1 flows do not use this cache.
+ */
+public class PacketIdCache {
+
+ private DuplicateIDCache cache;
+ private MQTTSession session;
+ private PostOffice postOffice;
+ private final SimpleString cacheName;
+
+ public PacketIdCache(MQTTSession session, TYPE type) {
+ this.session = session;
+ this.postOffice = session.getServer().getPostOffice();
+ this.cacheName = getCacheName(session.getServer().getInternalNamingPrefix(), session.getState().getClientId(), type);
+ }
+
+ /*
+ * "Getting" the cache from the PostOffice automatically creates it if it doesn't exist. We want to avoid this for
+ * most operations.
+ */
+ private void check() {
+ if (cache == null && durableStateExists()) {
+ cache = postOffice.getDuplicateIDCache(cacheName, MQTTUtil.TWO_BYTE_INT_MAX);
+ }
+ }
+
+ private boolean durableStateExists() {
+ return postOffice.duplicateIDCacheExists(cacheName);
+ }
+
+ public void add(int packetId, Transaction tx) throws Exception {
+ if (cache == null) {
+ cache = postOffice.getDuplicateIDCache(cacheName, MQTTUtil.TWO_BYTE_INT_MAX);
+ }
+ cache.addToCache(ByteUtil.intToBytes(packetId), tx);
+ }
+
+ public boolean contains(int packetId) {
+ check();
+ return cache != null && cache.contains(ByteUtil.intToBytes(packetId));
+ }
+
+ public boolean remove(int packetId) throws Exception {
+ check();
+ return cache != null && cache.deleteFromCache(ByteUtil.intToBytes(packetId));
+ }
+
+ public int size() {
+ check();
+ return cache != null ? cache.getMap().size() : 0;
+ }
+
+ public List<Integer> getPacketIds() {
+ check();
+ if (cache == null) {
+ return Collections.emptyList();
+ }
+ List<Integer> result = new ArrayList<>();
+ for (Pair<byte[], Long> entry : cache.getMap()) {
+ result.add(ByteUtil.bytesToInt(entry.getA()));
+ }
+ return result;
+ }
+
+ public void clear() throws Exception {
+ postOffice.deleteDuplicateCache(cacheName);
+ cache = null;
+ }
+
+ public static SimpleString getCacheName(String prefix, String clientId, TYPE type) {
+ return SimpleString.of(prefix).concat("mqtt.qos.").concat(type.type).concat('.').concat(clientId);
+ }
+
+ public enum TYPE {
+ PUBLISH(SimpleString.of("publish")),
+ PUBREC(SimpleString.of("pubrec"));
+
+ final SimpleString type;
+
+ TYPE(SimpleString type) {
+ this.type = type;
+ }
+ }
+}
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/PacketIdCorrelationKey.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/PacketIdCorrelationKey.java
new file mode 100644
index 0000000000..6aafb6379e
--- /dev/null
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/PacketIdCorrelationKey.java
@@ -0,0 +1,133 @@
+/*
+ * 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.core.protocol.mqtt;
+
+import java.util.Objects;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.journal.collections.AbstractHashMapPersister;
+import org.apache.activemq.artemis.utils.BufferHelper;
+import org.apache.activemq.artemis.utils.DataConstants;
+
+/**
+ * Composite key that maps a core message delivery to its MQTT packet ID. The MQTT specification requires that packet
+ * IDs for in-flight QoS 1 and QoS 2 messages are unique per client session, and that the same packet ID is reused if
+ * the message is redelivered (e.g. after a reconnect). This mapping is persisted in the journal so that a reconnecting
+ * client receives the same packet ID it was originally assigned. The key includes both the core message ID and the
+ * subscription address because overlapping subscriptions (e.g. {@code foo/bar} and {@code foo/#}) can cause the same
+ * core message to be delivered to the same client more than once, each through a different subscription address and
+ * with its own packet ID.
+ */
+public class PacketIdCorrelationKey {
+
+ private static Persister persister = new Persister();
+
+ public static Persister getPersister() {
+ return persister;
+ }
+
+ private long coreMessageId;
+ private SimpleString address;
+
+ public static PacketIdCorrelationKey of(long coreMessageId, SimpleString address) {
+ return new PacketIdCorrelationKey(coreMessageId, address);
+ }
+
+ private PacketIdCorrelationKey(long coreMessageId, SimpleString address) {
+ this.coreMessageId = coreMessageId;
+ this.address = address;
+ }
+
+ public long getCoreMessageId() {
+ return coreMessageId;
+ }
+
+ public SimpleString getAddress() {
+ return address;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof PacketIdCorrelationKey other)) {
+ return false;
+ }
+ return coreMessageId == other.coreMessageId &&
+ Objects.equals(address, other.address);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(coreMessageId, address);
+ }
+
+ @Override
+ public String toString() {
+ return "PacketIdCorrelation[" + "coreMessageId=" + coreMessageId + ", address=" + address + "]";
+ }
+
+ private static class Persister extends AbstractHashMapPersister<String, PacketIdCorrelationKey, Integer> {
+ @Override
+ protected int getCollectionIdSize(String collectionID) {
+ return BufferHelper.sizeOfString(collectionID);
+ }
+
+ @Override
+ protected void encodeCollectionId(ActiveMQBuffer buffer, String collectionID) {
+ buffer.writeString(collectionID);
+ }
+
+ @Override
+ protected String decodeCollectionId(ActiveMQBuffer buffer) {
+ return buffer.readString();
+ }
+
+ @Override
+ protected int getKeySize(PacketIdCorrelationKey packetIdCorrelationKey) {
+ return DataConstants.SIZE_LONG + packetIdCorrelationKey.getAddress().sizeof();
+ }
+
+ @Override
+ protected void encodeKey(ActiveMQBuffer buffer, PacketIdCorrelationKey packetIdCorrelationKey) {
+ buffer.writeLong(packetIdCorrelationKey.getCoreMessageId());
+ buffer.writeSimpleString(packetIdCorrelationKey.getAddress());
+ }
+
+ @Override
+ protected PacketIdCorrelationKey decodeKey(ActiveMQBuffer buffer) {
+ return PacketIdCorrelationKey.of(buffer.readLong(), buffer.readSimpleString());
+ }
+
+ @Override
+ protected int getValueSize(Integer mqttPacketId) {
+ return DataConstants.SIZE_INT;
+ }
+
+ @Override
+ protected void encodeValue(ActiveMQBuffer buffer, Integer mqttPacketId) {
+ buffer.writeInt(mqttPacketId);
+ }
+
+ @Override
+ protected Integer decodeValue(ActiveMQBuffer buffer, PacketIdCorrelationKey coreMessageId) {
+ return buffer.readInt();
+ }
+ }
+}
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/SubscriptionItem.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/SubscriptionItem.java
new file mode 100644
index 0000000000..1f07450ef0
--- /dev/null
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/SubscriptionItem.java
@@ -0,0 +1,83 @@
+/*
+ * 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.core.protocol.mqtt;
+
+import io.netty.handler.codec.mqtt.MqttTopicSubscription;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.postoffice.Address;
+import org.apache.activemq.artemis.core.postoffice.impl.AddressImpl;
+import org.apache.activemq.artemis.core.server.ServerConsumer;
+
+public class SubscriptionItem {
+
+ private MqttTopicSubscription subscription;
+ private Integer id;
+ private Address address;
+ private volatile ServerConsumer consumer;
+
+ public static SubscriptionItem of(MqttTopicSubscription subscription, Integer id) {
+ return new SubscriptionItem(subscription, id, null);
+ }
+
+ public static SubscriptionItem of(MqttTopicSubscription subscription, Integer id, ServerConsumer consumer) {
+ return new SubscriptionItem(subscription, id, consumer);
+ }
+
+ private SubscriptionItem(MqttTopicSubscription subscription, Integer id, ServerConsumer consumer) {
+ update(subscription, id);
+ this.consumer = consumer;
+ }
+
+ public MqttTopicSubscription getSubscription() {
+ return subscription;
+ }
+
+ public Integer getId() {
+ return id;
+ }
+
+ public ServerConsumer getConsumer() {
+ return consumer;
+ }
+
+ public SubscriptionItem setConsumer(ServerConsumer consumer) {
+ this.consumer = consumer;
+ return this;
+ }
+
+ public Integer getMatchingId(Address topicToMatch) {
+ if (id != null && topicToMatch.matches(address)) {
+ return id;
+ } else {
+ return null;
+ }
+ }
+
+ public void update(MqttTopicSubscription newSub, Integer newId) {
+ if (newId != null && !newId.equals(id)) {
+ if (this.address == null || !subscription.topicFilter().equals(newSub.topicFilter())) {
+ String topicFilter = newSub.topicFilter();
+ if (MQTTUtil.isSharedSubscription(topicFilter)) {
+ topicFilter = MQTTUtil.decomposeSharedSubscriptionTopicFilter(newSub.topicFilter()).getB();
+ }
+ address = new AddressImpl(SimpleString.of(topicFilter), MQTTUtil.MQTT_WILDCARD);
+ }
+ }
+ subscription = newSub;
+ id = newId;
+ }
+}
\ No newline at end of file
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionStateTest.java b/artemis-protocols/artemis-mqtt-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionStateTest.java
index 7e37bd3e9a..6b6eaa16c4 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionStateTest.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionStateTest.java
@@ -30,8 +30,8 @@ public class MQTTSessionStateTest {
public void testGenerateMqttIdOrder() {
MQTTSessionState mqttSessionState = new MQTTSessionState(RandomUtil.randomUUIDString());
for (int i = 1; i <= MQTTUtil.TWO_BYTE_INT_MAX; i++) {
- assertEquals(i, mqttSessionState.getOutboundStore().generateMqttId(RandomUtil.randomLong(), RandomUtil.randomLong()));
- mqttSessionState.getOutboundStore().publish(i, RandomUtil.randomLong(), RandomUtil.randomLong());
+ assertEquals(i, mqttSessionState.generatePacketId());
+ mqttSessionState.putCoreDeliveryInfo(i, CoreDeliveryInfo.of(RandomUtil.randomLong(), PacketIdCorrelationKey.of(RandomUtil.randomLong(), RandomUtil.randomUUIDSimpleString())));
}
}
@@ -41,11 +41,11 @@ public class MQTTSessionStateTest {
for (int i = 0; i < 10_000; i++) {
int random = RandomUtil.randomInterval(1, MQTTUtil.TWO_BYTE_INT_MAX);
// acknowledge a random ID
- mqttSessionState.getOutboundStore().publishAckd(random);
+ mqttSessionState.removeCoreDeliveryInfo(random);
// the ID that was acked is now the only one available so ensure generator finds it
- assertEquals(random, mqttSessionState.getOutboundStore().generateMqttId(RandomUtil.randomLong(), RandomUtil.randomLong()));
+ assertEquals(random, mqttSessionState.generatePacketId());
// "publish" a new message with the ID to ensure the store is full for the next loop
- mqttSessionState.getOutboundStore().publish(random, RandomUtil.randomLong(), RandomUtil.randomLong());
+ mqttSessionState.putCoreDeliveryInfo(random, CoreDeliveryInfo.of(RandomUtil.randomLong(), PacketIdCorrelationKey.of(RandomUtil.randomLong(), RandomUtil.randomUUIDSimpleString())));
}
}
@@ -53,13 +53,13 @@ public class MQTTSessionStateTest {
public void testGenerateMqttIdExhausted() {
MQTTSessionState mqttSessionState = createMqttSessionStateWithFullOutboundStore();
// ensure we throw an exception when we try generate a new ID once the store is full
- assertThrows(IllegalStateException.class, () -> mqttSessionState.getOutboundStore().generateMqttId(RandomUtil.randomLong(), RandomUtil.randomLong()));
+ assertThrows(IllegalStateException.class, () -> mqttSessionState.generatePacketId());
}
private static MQTTSessionState createMqttSessionStateWithFullOutboundStore() {
MQTTSessionState mqttSessionState = new MQTTSessionState(RandomUtil.randomUUIDString());
for (int i = 1; i <= MQTTUtil.TWO_BYTE_INT_MAX; i++) {
- mqttSessionState.getOutboundStore().publish(i, RandomUtil.randomLong(), RandomUtil.randomLong());
+ mqttSessionState.putCoreDeliveryInfo(i, CoreDeliveryInfo.of(RandomUtil.randomLong(), PacketIdCorrelationKey.of(RandomUtil.randomLong(), RandomUtil.randomUUIDSimpleString())));
}
return mqttSessionState;
}
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTStateManagerTest.java b/artemis-protocols/artemis-mqtt-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTStateManagerTest.java
index 0b4b461ce7..21fdcc5b2d 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTStateManagerTest.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTStateManagerTest.java
@@ -26,6 +26,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.activemq.artemis.core.config.Configuration;
+import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
@@ -43,8 +44,10 @@ public class MQTTStateManagerTest {
public void testGetSessionStateNeverReturnsNullUnderConcurrentRemoval() throws Exception {
final ActiveMQServer server = mock(ActiveMQServer.class);
final Configuration configuration = mock(Configuration.class);
+ final StorageManager storageManager = mock(StorageManager.class);
when(server.getConfiguration()).thenReturn(configuration);
when(configuration.isMqttSubscriptionPersistenceEnabled()).thenReturn(false);
+ when(server.getStorageManager()).thenReturn(storageManager);
final MQTTStateManager manager = MQTTStateManager.getInstance(server);
try {
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/mqtt/StateSerDeTest.java b/artemis-protocols/artemis-mqtt-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/mqtt/StateSerDeTest.java
index 1d5630e99a..d82dd1501d 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/mqtt/StateSerDeTest.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/mqtt/StateSerDeTest.java
@@ -44,17 +44,17 @@ public class StateSerDeTest {
RandomUtil.randomBoolean(),
RandomUtil.randomBoolean(),
MqttSubscriptionOption.RetainedHandlingPolicy.valueOf(RandomUtil.randomInterval(0, 3))));
- unserialized.addSubscription(sub, MQTTUtil.MQTT_WILDCARD, subscriptionIdentifier);
+ unserialized.addSubscription(SubscriptionItem.of(sub, subscriptionIdentifier));
}
CoreMessage serializedState = MQTTStateManager.serializeState(unserialized, 0);
MQTTSessionState deserialized = new MQTTSessionState(serializedState);
assertEquals(unserialized.getClientId(), deserialized.getClientId());
- for (MQTTSessionState.SubscriptionItem unserializedItem : unserialized.getSubscriptionsPlusID().values()) {
+ for (SubscriptionItem unserializedItem : unserialized.getSubscriptionsPlusID().values()) {
MqttTopicSubscription unserializedSub = unserializedItem.getSubscription();
Integer unserializedSubId = unserializedItem.getId();
- MQTTSessionState.SubscriptionItem deserializedEntry = deserialized.getSubscriptionPlusID(unserializedSub.topicFilter());
+ SubscriptionItem deserializedEntry = deserialized.getSubscriptionItem(unserializedSub.topicFilter());
MqttTopicSubscription deserializedSub = deserializedEntry.getSubscription();
Integer deserializedSubId = deserializedEntry.getId();
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalRecordIds.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalRecordIds.java
index b29ca6efe7..29530ebb71 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalRecordIds.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalRecordIds.java
@@ -105,4 +105,6 @@ public final class JournalRecordIds {
public static final byte ADDRESS_SETTING_RECORD_JSON = 52;
public static final byte ACK_RETRY = 53;
+
+ public static final byte MQTT_PACKET_ID_CORRELATION = 54;
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/DuplicateIDCache.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/DuplicateIDCache.java
index 2f0d3269a8..9ed47e6923 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/DuplicateIDCache.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/DuplicateIDCache.java
@@ -39,7 +39,7 @@ public interface DuplicateIDCache {
*/
void addToCache(byte[] duplicateID, Transaction tx, boolean instantAdd) throws Exception;
- void deleteFromCache(byte[] duplicateID) throws Exception;
+ boolean deleteFromCache(byte[] duplicateID) throws Exception;
void load(List<Pair<byte[], Long>> ids) throws Exception;
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java
index f8ecb42f0d..45c29671da 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/PostOffice.java
@@ -191,10 +191,14 @@ public interface PostOffice extends ActiveMQComponent {
void processRoute(Message message, RoutingContext context, boolean direct) throws Exception;
+ boolean duplicateIDCacheExists(SimpleString address);
+
DuplicateIDCache getDuplicateIDCache(SimpleString address);
DuplicateIDCache getDuplicateIDCache(SimpleString address, int idCacheSize);
+ void deleteDuplicateCache(SimpleString address) throws Exception;
+
void sendQueueInfoToQueue(SimpleString queueName, SimpleString address) throws Exception;
Object getNotificationLock();
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/InMemoryDuplicateIDCache.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/InMemoryDuplicateIDCache.java
index dccce30a31..c35e461800 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/InMemoryDuplicateIDCache.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/InMemoryDuplicateIDCache.java
@@ -77,11 +77,11 @@ final class InMemoryDuplicateIDCache implements DuplicateIDCache {
}
@Override
- public void deleteFromCache(byte[] duplicateID) {
- deleteFromCache(new ByteArray(duplicateID));
+ public boolean deleteFromCache(byte[] duplicateID) {
+ return deleteFromCache(new ByteArray(duplicateID));
}
- private void deleteFromCache(final ByteArray duplicateID) {
+ private boolean deleteFromCache(final ByteArray duplicateID) {
if (logger.isTraceEnabled()) {
logger.trace("deleting id = {}", describeID(duplicateID.bytes));
}
@@ -102,8 +102,10 @@ final class InMemoryDuplicateIDCache implements DuplicateIDCache {
}
}
}
+ return true;
}
+ return false;
}
private static String describeID(byte[] duplicateID) {
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/NoOpDuplicateIDCache.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/NoOpDuplicateIDCache.java
index 5443de2b2f..9190487e38 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/NoOpDuplicateIDCache.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/NoOpDuplicateIDCache.java
@@ -51,8 +51,8 @@ public final class NoOpDuplicateIDCache implements DuplicateIDCache {
}
@Override
- public void deleteFromCache(byte[] duplicateID) throws Exception {
-
+ public boolean deleteFromCache(byte[] duplicateID) throws Exception {
+ return true;
}
@Override
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PersistentDuplicateIDCache.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PersistentDuplicateIDCache.java
index bcccddfa46..acb202395e 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PersistentDuplicateIDCache.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PersistentDuplicateIDCache.java
@@ -145,11 +145,11 @@ final class PersistentDuplicateIDCache implements DuplicateIDCache {
}
@Override
- public void deleteFromCache(byte[] duplicateID) throws Exception {
- deleteFromCache(new ByteArray(duplicateID));
+ public boolean deleteFromCache(byte[] duplicateID) throws Exception {
+ return deleteFromCache(new ByteArray(duplicateID));
}
- private void deleteFromCache(final ByteArray duplicateID) throws Exception {
+ private boolean deleteFromCache(final ByteArray duplicateID) throws Exception {
if (logger.isTraceEnabled()) {
logger.trace("deleting id = {}", describeID(duplicateID.bytes));
}
@@ -170,8 +170,10 @@ final class PersistentDuplicateIDCache implements DuplicateIDCache {
storageManager.deleteDuplicateID(recordID);
}
}
+ return true;
}
+ return false;
}
private static String describeID(byte[] duplicateID) {
@@ -322,15 +324,19 @@ final class PersistentDuplicateIDCache implements DuplicateIDCache {
logger.debug("address = {} removing duplicate ID data", address);
final int idsSize = ids.size();
if (idsSize > 0) {
+ boolean deleted = false;
long tx = storageManager.generateID();
for (int i = 0; i < idsSize; i++) {
final ObjLongPair<ByteArray> id = ids.get(i);
if (id.getA() != null) {
assert id.getB() != NIL;
storageManager.deleteDuplicateIDTransactional(tx, id.getB());
+ deleted = true;
}
}
- storageManager.commit(tx);
+ if (deleted) {
+ storageManager.commit(tx);
+ }
}
ids.clear();
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
index 6ee9a03a1a..1dd2ef5b42 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
@@ -1051,7 +1051,8 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
}
}
- private void deleteDuplicateCache(SimpleString address) throws Exception {
+ @Override
+ public void deleteDuplicateCache(SimpleString address) throws Exception {
DuplicateIDCache cache = duplicateIDCaches.remove(address);
if (cache != null) {
@@ -1476,6 +1477,11 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
return Objects.requireNonNullElse(addressSettingsRepository.getMatch(address.toString()).getIDCacheSize(), idCacheSize);
}
+ @Override
+ public boolean duplicateIDCacheExists(final SimpleString address) {
+ return duplicateIDCaches.containsKey(address);
+ }
+
@Override
public DuplicateIDCache getDuplicateIDCache(final SimpleString address) {
int resolvedIdCacheSize = resolveIdCacheSize(address);
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptor.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptor.java
index 0d258081af..44dcb821a0 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptor.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptor.java
@@ -180,7 +180,7 @@ public final class InVMAcceptor extends AbstractAcceptor {
try {
notificationService.sendNotification(notification);
} catch (Exception e) {
- ActiveMQServerLogger.LOGGER.failedToSendNotification(e);
+ ActiveMQServerLogger.LOGGER.failedToSendNotification(notification.toString(), e.getMessage());
}
}
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
index 7943f4e197..1bd19470e8 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
@@ -900,7 +900,7 @@ public class NettyAcceptor extends AbstractAcceptor {
try {
notificationService.sendNotification(notification);
} catch (Exception e) {
- ActiveMQServerLogger.LOGGER.failedToSendNotification(e);
+ ActiveMQServerLogger.LOGGER.failedToSendNotification(notification.toString(), e.getMessage());
}
}
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/RemotingService.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/RemotingService.java
index 652c89fd86..d1e0a3cf7a 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/RemotingService.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/RemotingService.java
@@ -75,6 +75,8 @@ public interface RemotingService {
boolean removeOutgoingInterceptor(BaseInterceptor interceptor);
+ void clearInterceptors();
+
void notifyStop();
/** The Prepare stop will close all the connections however it will use the one used by storage manager */
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/impl/RemotingServiceImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/impl/RemotingServiceImpl.java
index f367ddf2d4..ef30d9b406 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/impl/RemotingServiceImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/impl/RemotingServiceImpl.java
@@ -608,7 +608,11 @@ public class RemotingServiceImpl implements RemotingService, ServerConnectionLif
@Override
public void loadProtocolServices(List<ActiveMQComponent> protocolServices) {
for (ProtocolManagerFactory protocolManagerFactory : protocolMap.values()) {
- protocolManagerFactory.loadProtocolServices(this.server, protocolServices);
+ try {
+ protocolManagerFactory.loadProtocolServices(this.server, protocolServices);
+ } catch (Exception e) {
+ logger.warn("Unable to load protocol services for: {}", protocolManagerFactory.getProtocols(), e);
+ }
}
}
@@ -880,6 +884,13 @@ public class RemotingServiceImpl implements RemotingService, ServerConnectionLif
}
}
+ @Override
+ public void clearInterceptors() {
+ outgoingInterceptors.clear();
+ incomingInterceptors.clear();
+ updateProtocols();
+ }
+
private ClusterConnection lookupClusterConnection(TransportConfiguration acceptorConfig) {
String clusterConnectionName = (String) acceptorConfig.getParams().get(org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.CLUSTER_CONNECTION);
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java
index 0ca2902c56..35cf342bc0 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQMessageBundle.java
@@ -130,7 +130,7 @@ public interface ActiveMQMessageBundle {
@Message(id = 229026, value = "Backup Server was not yet in sync with live")
ActiveMQIllegalStateException backupServerNotInSync();
- @Message(id = 229027, value = "Could not find reference on consumer ID={}, messageId = {} queue = {}")
+ @Message(id = 229027, value = "Could not find reference on consumerId = {}, messageId = {} queue = {}")
ActiveMQIllegalStateException consumerNoReference(Long id, Long messageID, SimpleString name);
@Message(id = 229028, value = "Consumer {} doesn't exist on the server")
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
index 4f33690077..f27e0df700 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
@@ -882,8 +882,8 @@ public interface ActiveMQServerLogger {
@LogMessage(id = 222229, value = "Failed to perform rollback", level = LogMessage.Level.WARN)
void failedToPerformRollback(IllegalStateException e);
- @LogMessage(id = 222230, value = "Failed to send notification", level = LogMessage.Level.WARN)
- void failedToSendNotification(Exception e);
+ @LogMessage(id = 222230, value = "Failed to send notification: {}; Exception message: {}", level = LogMessage.Level.WARN)
+ void failedToSendNotification(String notification, String exceptionMessage);
@LogMessage(id = 222231, value = "Failed to flush outstanding data from the connection", level = LogMessage.Level.WARN)
void failedToFlushOutstandingDataFromTheConnection(Throwable e);
@@ -1550,4 +1550,10 @@ public interface ActiveMQServerLogger {
@LogMessage(id = 224164, value = "Failed to recover stored configuration for divert named '{}': {}. To repair this record create a new divert with the same name via the management API.", level = LogMessage.Level.WARN)
void failedToRecoverStoredDivertConfiguration(String divertName, String divert);
+
+ @LogMessage(id = 224165, value = "Server is stopping. Unable to process redelivery during rollback; ref: {}; transaction: {}; exception message: {}", level = LogMessage.Level.WARN)
+ void unableToProcessRedeliveryDuringRollback(String messageRef, String transaction, String exceptionMessage);
+
+ @LogMessage(id = 224166, value = "Server is stopping. Unable to delete unreferenced message with id={}.", level = LogMessage.Level.WARN)
+ void unableToDeleteMessageDuringShutdown(long messageId);
}
\ No newline at end of file
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
index ee7e22817a..049028ee8f 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
@@ -3711,7 +3711,11 @@ public class ActiveMQServerImpl implements ActiveMQServer {
private void startProtocolServices() throws Exception {
for (ProtocolManagerFactory protocolManagerFactory : protocolManagerFactories) {
- protocolManagerFactory.loadProtocolServices(this, protocolServices);
+ try {
+ protocolManagerFactory.loadProtocolServices(this, protocolServices);
+ } catch (Exception e) {
+ logger.warn("Unable to load protocol services for: {}", protocolManagerFactory.getProtocols(), e);
+ }
}
for (ActiveMQComponent protocolComponent : protocolServices) {
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
index 4d70d3f6cf..9b23fd7e48 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
@@ -49,6 +49,7 @@ import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.ActiveMQNullRefException;
import org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException;
+import org.apache.activemq.artemis.api.core.ActiveMQShutdownException;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.Pair;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
@@ -4021,6 +4022,8 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
// There is a startup check to remove non referenced messages case these deletes fail
try {
storageManager.deleteMessage(message.getMessageID());
+ } catch (ActiveMQShutdownException e) {
+ ActiveMQServerLogger.LOGGER.unableToDeleteMessageDuringShutdown(message.getMessageID());
} catch (Exception e) {
ActiveMQServerLogger.LOGGER.cannotFindMessageOnJournal(message.getMessageID(), e);
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java
index 8ef9afffca..800bd65a86 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java
@@ -22,6 +22,7 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import org.apache.activemq.artemis.api.core.ActiveMQShutdownException;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.RefCountMessage;
import org.apache.activemq.artemis.core.paging.cursor.PagedReference;
@@ -117,6 +118,8 @@ public class RefsOperation extends TransactionOperationAbstract {
ackedRefs.add(ref);
}
rollbackRedelivery(tx, ref, timeBase, queueMap);
+ } catch (ActiveMQShutdownException e) {
+ ActiveMQServerLogger.LOGGER.unableToProcessRedeliveryDuringRollback(ref.toString(), tx.toString(), e.getMessage());
} catch (Exception e) {
ActiveMQServerLogger.LOGGER.errorCheckingDLQ(e);
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
index 01764ae8e5..be1940f735 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
@@ -513,10 +513,11 @@ public class ServerSessionImpl extends CriticalComponentImpl implements ServerSe
props.putSimpleStringProperty(ManagementHelper.HDR_PROTOCOL_NAME, SimpleString.of(stableRemotingConnection.getProtocolName()));
props.putSimpleStringProperty(ManagementHelper.HDR_ADDRESS, managementService.getManagementNotificationAddress());
props.putIntProperty(ManagementHelper.HDR_DISTANCE, 0);
+ Notification notification = new Notification(null, type, props);
try {
- managementService.sendNotification(new Notification(null, type, props));
+ managementService.sendNotification(notification);
} catch (Exception e) {
- ActiveMQServerLogger.LOGGER.failedToSendNotification(e);
+ ActiveMQServerLogger.LOGGER.failedToSendNotification(notification.toString(), e.getMessage());
}
});
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractProtocolManagerFactory.java b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractProtocolManagerFactory.java
index 96be40f880..955ed86dff 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractProtocolManagerFactory.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractProtocolManagerFactory.java
@@ -46,7 +46,7 @@ public abstract class AbstractProtocolManagerFactory<P extends BaseInterceptor>
}
@Override
- public void loadProtocolServices(ActiveMQServer server, List<ActiveMQComponent> services) {
+ public void loadProtocolServices(ActiveMQServer server, List<ActiveMQComponent> services) throws Exception {
}
@Override
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManagerFactory.java b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManagerFactory.java
index d61f0a09ad..b16f7e3460 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManagerFactory.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManagerFactory.java
@@ -50,7 +50,7 @@ public interface ProtocolManagerFactory<P extends BaseInterceptor> {
String getModuleName();
- void loadProtocolServices(ActiveMQServer server, List<ActiveMQComponent> services);
+ void loadProtocolServices(ActiveMQServer server, List<ActiveMQComponent> services) throws Exception;
/**
* Provides an entry point for the server to trigger the protocol manager factory to update its protocol services
diff --git a/docs/user-manual/versions.adoc b/docs/user-manual/versions.adoc
index 3b6653a777..c1259f303b 100644
--- a/docs/user-manual/versions.adoc
+++ b/docs/user-manual/versions.adoc
@@ -28,6 +28,18 @@ https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315920&versio
* Due to https://issues.apache.org/jira/browse/ARTEMIS-6128[ARTEMIS-6128] the `Alder32` and `fileAlder32` properties have been deprecated in favor of `Adler32` and `fileAdler32` respectively.
* Due to https://issues.apache.org/jira/browse/ARTEMIS-6172[ARTEMIS-6172] the HTTP-specific transport parameters deprecated by https://issues.apache.org/jira/browse/ARTEMIS-5819[ARTEMIS-5819] (`httpClientIdleTime`, `httpClientIdleScanPeriod`, `httpResponseTime`, `httpServerScanPeriod`) are restored.
These are required so HTTP-tunneled Core connections can create response slots for server-initiated packets (for example cluster topology updates) without starving replies to client requests.
+* Due to https://issues.apache.org/jira/browse/ARTEMIS-6189[ARTEMIS-6189] MQTT QoS 1 and QoS 2 message delivery has been redesigned for resiliency across broker restarts and client reconnects.
+Additional persistent state is now tracked via duplicate ID caches (named `mqtt.qos.publish.<clientId>` and `mqtt.qos.pubrec.<clientId>`) and packet ID correlation records.
+These are internal implementation details, but they will be visible in data tool output.
++
+The per-client internal management queues previously used for QoS 2 PUBREL persistence (named `$sys.mqtt.queue.qos2.<clientId>`) are no longer used.
+It is recommended to allow all in-flight QoS 1 and QoS 2 flows to complete before upgrading so that QoS flow state information is not orphaned in the existing management queues.
+Manual clean-up of these queues may be required after upgrading.
++
+Due to the additional persistent state there will be a performance impact.
+Of course, this is a traditional trade-off for increased reliability.
+The exact impact will depend on your specific use-case.
+Generally speaking, the recommendation is to use the lowest QoS possible to avoid unnecessary overhead.
== Version 2.55.0
diff --git a/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakePostOffice.java b/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakePostOffice.java
index 66e7a36389..63b28a5c4b 100644
--- a/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakePostOffice.java
+++ b/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/tests/unit/core/server/impl/fakes/FakePostOffice.java
@@ -204,6 +204,10 @@ public class FakePostOffice implements PostOffice {
return DuplicateIDCaches.inMemory(address, idSize);
}
+ @Override
+ public void deleteDuplicateCache(SimpleString address) throws Exception {
+ }
+
@Override
public Collection<Binding> getMatchingBindings(final SimpleString address) {
@@ -293,6 +297,11 @@ public class FakePostOffice implements PostOffice {
public void processRoute(Message message, RoutingContext context, boolean direct) throws Exception {
}
+ @Override
+ public boolean duplicateIDCacheExists(SimpleString address) {
+ return false;
+ }
+
@Override
public RoutingStatus route(Message message, boolean direct) throws Exception {
return RoutingStatus.OK;
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/MQTTTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/MQTTTest.java
index 03cbcb7a1e..58b4c4ac18 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/MQTTTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/MQTTTest.java
@@ -62,7 +62,6 @@ import org.apache.activemq.artemis.core.postoffice.Binding;
import org.apache.activemq.artemis.core.postoffice.QueueBinding;
import org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
-import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.json.JsonArray;
@@ -100,7 +99,6 @@ public class MQTTTest extends MQTTTestSupport {
private static final String AMQP_URI = "tcp://localhost:61616";
-
@Override
public void configureBroker() throws Exception {
super.configureBroker();
@@ -320,30 +318,6 @@ public class MQTTTest extends MQTTTestSupport {
provider.disconnect();
}
- @Test
- @Timeout(120)
- public void testManagementQueueMessagesAreAckd() throws Exception {
- String clientId = "test.client.id";
- final MQTTClientProvider provider = getMQTTClientProvider();
- provider.setClientId(clientId);
- initializeConnection(provider);
- provider.subscribe("foo", EXACTLY_ONCE);
- for (int i = 0; i < NUM_MESSAGES; i++) {
- String payload = "Test Message: " + i;
- provider.publish("foo", payload.getBytes(), EXACTLY_ONCE);
- byte[] message = provider.receive(5000);
- assertNotNull(message, "Should get a message");
- assertEquals(payload, new String(message));
- }
-
- final Queue queue = server.locateQueue(SimpleString.of(MQTTUtil.QOS2_MANAGEMENT_QUEUE_PREFIX + clientId));
-
- Wait.waitFor(() -> queue.getMessageCount() == 0, 1000, 100);
-
- assertEquals(0, queue.getMessageCount());
- provider.disconnect();
- }
-
@Test
@Timeout(120)
public void testSendAtLeastOnceReceiveExactlyOnce() throws Exception {
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/PahoMQTTQOS2SecurityTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/PahoMQTTQOS2SecurityTest.java
index f012221764..628be58f0a 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/PahoMQTTQOS2SecurityTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/PahoMQTTQOS2SecurityTest.java
@@ -16,8 +16,9 @@
*/
package org.apache.activemq.artemis.tests.integration.mqtt;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
import org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil;
import org.apache.activemq.artemis.core.security.Role;
@@ -34,9 +35,8 @@ import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.concurrent.CountDownLatch;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
public class PahoMQTTQOS2SecurityTest extends MQTTTestSupport {
@@ -134,7 +134,7 @@ public class PahoMQTTQOS2SecurityTest extends MQTTTestSupport {
} catch (MqttException e) {
// ignore
}
- assertEquals(0, getSessions().get(clientID).getPubRec().size());
+ assertEquals(0, getSessions().get(clientID).getPublishCache().size());
producer.close();
}
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 dcae8935a2..ff104964b1 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
@@ -913,13 +913,13 @@ public class MQTT5Test extends MQTT5TestSupport {
// ensure subscriber got message and ack was blocked
assertTrue(subscriberLatch.await(500, TimeUnit.MILLISECONDS));
assertTrue(interceptorBlockedLatch.await(500, TimeUnit.MILLISECONDS));
- Wait.assertEquals(1L, () -> mqttSessionState.getOutboundStore().getSendQuota(), 2000, 10);
+ Wait.assertEquals(1L, () -> mqttSessionState.getSendQuota(), 2000, 10);
pendingCountCheckLatch.countDown();
// disconnect subscriber
subscriber.disconnect();
Wait.assertFalse(() -> mqttSessionState.isAttached(), 2000, 50);
- assertEquals(0, mqttSessionState.getOutboundStore().getSendQuota());
+ assertEquals(0, mqttSessionState.getSendQuota());
assertEquals(1L, subscriptionQueue.getMessageCount());
assertEquals(0L, subscriptionQueue.getMessagesAcknowledged());
assertEquals(0L, subscriptionQueue.getConsumerCount());
@@ -1002,6 +1002,65 @@ public class MQTT5Test extends MQTT5TestSupport {
assertEquals(RoutingType.MULTICAST, getSubscriptionQueue(topic, clientID).getRoutingType());
}
+ /**
+ * This test stresses the synchronization between the broker sending the CONNACK and starting the SubscriptionManager
+ * and the client sending a SUBSCRIBE.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testConcurrentReconnectAndResubscribe() throws Exception {
+ final int clientCount = 100;
+ final int subsPerClient = 100;
+ final int resubscribeCount = 10;
+ AtomicBoolean failed = new AtomicBoolean(false);
+ ExecutorService executorService = Executors.newFixedThreadPool(clientCount);
+ runAfter(executorService::shutdownNow);
+ CountDownLatch latch = new CountDownLatch(clientCount);
+
+ for (int c = 0; c < clientCount; c++) {
+ final String clientId = "client-" + c;
+ final int clientIndex = c;
+ executorService.submit(() -> {
+ MqttClient client = null;
+ try {
+ client = createPahoClient(clientId);
+ MqttConnectionOptions options = new MqttConnectionOptionsBuilder()
+ .cleanStart(false)
+ .sessionExpiryInterval(300L)
+ .build();
+
+ MqttSubscription[] subs = new MqttSubscription[subsPerClient];
+ for (int s = 0; s < subsPerClient; s++) {
+ subs[s] = new MqttSubscription("topic/" + clientIndex + "/" + s, AT_LEAST_ONCE);
+ }
+
+ for (int i = 0; i < resubscribeCount; i++) {
+ client.connect(options);
+ client.subscribe(subs);
+ client.disconnect();
+ }
+ client.close();
+ } catch (Exception e) {
+ logger.error("Client {} failed: {}", clientId, e.getMessage(), e);
+ failed.set(true);
+ } finally {
+ if (client != null) {
+ try {
+ client.disconnect();
+ client.close();
+ } catch (MqttException e) {
+ // ignore
+ }
+ }
+ latch.countDown();
+ }
+ });
+ }
+
+ latch.await();
+ assertFalse(failed.get());
+ }
+
@Test
@Timeout(DEFAULT_TIMEOUT_SEC)
public void testPublishWithDelimiterInTopicNameAndWildcardSubscription() throws Exception {
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5TestSupport.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5TestSupport.java
index ce19a1e76e..bdc39ff06e 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5TestSupport.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5TestSupport.java
@@ -37,7 +37,9 @@ import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
+import org.apache.activemq.artemis.core.postoffice.DuplicateIDCache;
import org.apache.activemq.artemis.core.protocol.mqtt.MQTTInterceptor;
+import org.apache.activemq.artemis.core.protocol.mqtt.PacketIdCache;
import org.apache.activemq.artemis.core.protocol.mqtt.MQTTProtocolManager;
import org.apache.activemq.artemis.core.protocol.mqtt.MQTTSessionState;
import org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil;
@@ -55,6 +57,11 @@ import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
import org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.utils.ClassloadingUtil;
+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.eclipse.paho.mqttv5.client.IMqttToken;
import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
import org.eclipse.paho.mqttv5.client.MqttCallback;
@@ -73,6 +80,14 @@ import static java.util.Collections.singletonList;
import static org.apache.activemq.artemis.core.protocol.mqtt.MQTTProtocolManagerFactory.MQTT_PROTOCOL_NAME;
public class MQTT5TestSupport extends ActiveMQTestBase {
+
+ // The Paho MQTT client logging adds noise during QoS2 operations
+ private static final java.util.logging.Logger PAHO_LOGGER;
+ static {
+ PAHO_LOGGER = java.util.logging.Logger.getLogger("org.eclipse.paho.mqttv5.client.internal.ClientState");
+ PAHO_LOGGER.setLevel(java.util.logging.Level.WARNING);
+ }
+
protected static final String TCP = "tcp";
protected static final String WS = "ws";
protected static final String SSL = "ssl";
@@ -158,15 +173,36 @@ public class MQTT5TestSupport extends ActiveMQTestBase {
exceptions.clear();
startBroker();
createJMSConnection();
+ if (isProtocolLoggingEnabled()) {
+ enableProtocolLogging();
+ }
}
@Override
@AfterEach
public void tearDown() throws Exception {
stopBroker();
+ if (isProtocolLoggingEnabled()) {
+ disableProtocolLogging();
+ }
super.tearDown();
}
+ public 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();
+ }
+
+ public 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();
+ }
+
public void configureBroker() throws Exception {
super.setUp();
server = createServerForMQTT();
@@ -305,6 +341,10 @@ public class MQTT5TestSupport extends ActiveMQTestBase {
return false;
}
+ public boolean isProtocolLoggingEnabled() {
+ return false;
+ }
+
protected interface Task {
void run() throws Exception;
@@ -351,6 +391,46 @@ public class MQTT5TestSupport extends ActiveMQTestBase {
server.getRemotingService().createAcceptor(MQTT_PROTOCOL_NAME, "tcp://localhost:" + port + "?protocols=MQTT;" + property).start();
}
+ protected DuplicateIDCache getPubCache(String clientId) {
+ return getCache(clientId, PacketIdCache.TYPE.PUBLISH);
+ }
+
+ protected int getPubCacheSize(String clientId) {
+ return getCacheSize(getPubCache(clientId));
+ }
+
+ protected DuplicateIDCache getPubRecCache(String clientId) {
+ return getCache(clientId, PacketIdCache.TYPE.PUBREC);
+ }
+
+ protected int getPubRecCacheSize(String clientId) {
+ return getCacheSize(getPubRecCache(clientId));
+ }
+
+ private int getCacheSize(DuplicateIDCache cache) {
+ 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);
+ } else {
+ return null;
+ }
+ }
+
+ protected static void reconnectSafely(MqttClient subscriber) throws Exception {
+ Wait.waitFor(() -> {
+ try {
+ subscriber.reconnect();
+ return true;
+ } catch (MqttException e) {
+ return false;
+ }
+ });
+ }
+
/*
* From the Paho MQTT client's JavaDoc:
*
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/spec/controlpackets/ConnAckTests.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/spec/controlpackets/ConnAckTests.java
index 0960275cd9..b270b2e28c 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/spec/controlpackets/ConnAckTests.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/spec/controlpackets/ConnAckTests.java
@@ -151,6 +151,17 @@ public class ConnAckTests extends MQTT5TestSupport {
result = consumer.connectWithResult(options);
assertTrue(result.getSessionPresent());
assertTrue(getListOfCodes(result.getResponse().getReasonCodes()).contains(MqttReturnCode.RETURN_CODE_SUCCESS));
+ consumer.disconnect();
+
+ server.stop();
+ waitForServerToStop(server);
+ server.start();
+ waitForServerToStart(server);
+
+ result = consumer.connectWithResult(options);
+ assertTrue(result.getSessionPresent());
+ assertTrue(getListOfCodes(result.getResponse().getReasonCodes()).contains(MqttReturnCode.RETURN_CODE_SUCCESS));
+ consumer.disconnect();
}
/*
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 d236a56d88..e28614a31a 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
@@ -485,7 +485,35 @@ public class ConnectTests extends MQTT5TestSupport {
*/
@Test
@Timeout(DEFAULT_TIMEOUT_SEC)
- public void testMaxPacketSize() throws Exception {
+ public void testMaxPacketSizeQoS2() throws Exception {
+ testMaxPacketSize(2);
+ }
+
+ /*
+ * [MQTT-3.1.2-24] The Server MUST NOT send packets exceeding Maximum Packet Size to the Client.
+ *
+ * [MQTT-3.1.2-25] Where a Packet is too large to send, the Server MUST discard it without sending it and then behave
+ * as if it had completed sending that Application Message.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testMaxPacketSizeQoS1() throws Exception {
+ testMaxPacketSize(1);
+ }
+
+ /*
+ * [MQTT-3.1.2-24] The Server MUST NOT send packets exceeding Maximum Packet Size to the Client.
+ *
+ * [MQTT-3.1.2-25] Where a Packet is too large to send, the Server MUST discard it without sending it and then behave
+ * as if it had completed sending that Application Message.
+ */
+ @Test
+ @Timeout(DEFAULT_TIMEOUT_SEC)
+ public void testMaxPacketSizeQoS0() throws Exception {
+ testMaxPacketSize(0);
+ }
+
+ private void testMaxPacketSize(int qos) throws Exception {
final String CONSUMER_ID = RandomUtil.randomUUIDString();
final String TOPIC = this.getTopicName();
final long SIZE = 1500;
@@ -502,11 +530,11 @@ public class ConnectTests extends MQTT5TestSupport {
.build();
consumer.setCallback(new LatchedMqttCallback(latch));
consumer.connect(options);
- consumer.subscribe(TOPIC, 2);
+ consumer.subscribe(TOPIC, qos);
MqttClient producer = createPahoClient(RandomUtil.randomUUIDString());
producer.connect();
- producer.publish(TOPIC, bytes, 2, false);
+ producer.publish(TOPIC, bytes, qos, false);
producer.disconnect();
producer.close();
Wait.assertEquals(1L, () -> getSubscriptionQueue(TOPIC, CONSUMER_ID).getMessagesAdded(), 2000, 100);
@@ -516,6 +544,9 @@ public class ConnectTests extends MQTT5TestSupport {
// the broker should acknowledge the message since it exceeded the client's max packet size
Wait.assertEquals(1L, () -> getSubscriptionQueue(TOPIC, CONSUMER_ID).getMessagesAcknowledged(), 2000, 100);
+ Wait.assertEquals(0L, () -> getSubscriptionQueue(TOPIC, CONSUMER_ID).getMessageCount(), 2000, 100);
+ Wait.assertEquals(0, () -> getProtocolManager().getStateManager().getPacketIdCorrelationSize(CONSUMER_ID), 2000, 100);
+ Wait.assertNull(() -> getPubRecCache(CONSUMER_ID), 2000, 100);
consumer.disconnect();
consumer.close();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@artemis.apache.org
For additional commands, e-mail: commits-help@artemis.apache.org