From 835551b4eb81e68f829aed12e2c253980b1520b6 Mon Sep 17 00:00:00 2001 From: Justin Bertram Date: Mon, 5 Oct 2026 09:19:56 -0500 Subject: [PATCH] ARTEMIS-6284 make MQTT cache removal transactional Although this commit makes cache removal transactional the same effective semantic could potentially have been achieved by ensuring a sync happened on non-transactional removal. However, this change to use a transaction has undergone extensive testing and has proven to be resilient. --- .../protocol/mqtt/MQTTPublishManager.java | 22 ++++++++++++++----- .../core/protocol/mqtt/PacketIdCache.java | 4 ++-- .../core/postoffice/DuplicateIDCache.java | 2 ++ .../impl/InMemoryDuplicateIDCache.java | 5 +++++ .../postoffice/impl/NoOpDuplicateIDCache.java | 5 +++++ .../impl/PersistentDuplicateIDCache.java | 19 ++++++++++++---- 6 files changed, 46 insertions(+), 11 deletions(-) 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 5e6253e7efa..9e8b0f5d04f 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 @@ -325,17 +325,28 @@ private void releaseFlowControl(Long consumerId) { } synchronized void handlePubComp(int packetId) throws Exception { - session.getState().getPubRecCache().remove(packetId); + removePacketIdFromCache(session.getState().getPubRecCache(), packetId, MqttMessageType.PUBCOMP, MqttMessageType.PUBREC); } 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); - } + removePacketIdFromCache(session.getState().getPublishCache(), packetId, MqttMessageType.PUBREL, MqttMessageType.PUBLISH); session.getProtocolHandler().sendPubComp(packetId); } + private void removePacketIdFromCache(PacketIdCache cache, int packetId, MqttMessageType receivedType, MqttMessageType cachedType) throws Exception { + Transaction tx = session.getServerSession().newTransaction(); + try { + boolean deleted = cache.remove(packetId, tx); + if (!deleted) { + logger.debug("MQTT client {} sent {} for packet {} but no corresponding {} was found in the cache", session.getState().getClientId(), receivedType, packetId, cachedType); + } + tx.commit(); + } catch (Throwable t) { + tx.rollback(); + throw t; + } + } + synchronized void handlePubAck(int packetId) throws Exception { acknowledgeDelivery(packetId, false); } @@ -377,6 +388,7 @@ private void acknowledgeDelivery(int packetId, boolean needsPubRel) throws Excep state.getPubRecCache().add(packetId, tx); } session.getStateManager().removePacketIdCorrelation(state.getClientId(), delivery.getPacketIdCorrelationKey(), tx.getID()); + tx.setContainsPersistent(); consumer.individualAcknowledge(tx, delivery.getCoreMessageId()); tx.commit(); state.removeCoreDeliveryInfo(packetId); 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 index 70b29bbfd4e..7cc86d5f2f5 100644 --- 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 @@ -72,9 +72,9 @@ public boolean contains(int packetId) { return cache != null && cache.contains(ByteUtil.intToBytes(packetId)); } - public boolean remove(int packetId) throws Exception { + public boolean remove(int packetId, Transaction tx) throws Exception { check(); - return cache != null && cache.deleteFromCache(ByteUtil.intToBytes(packetId)); + return cache != null && cache.deleteFromCache(ByteUtil.intToBytes(packetId), tx); } public int size() { 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 9ed47e69237..79843376aa3 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 @@ -41,6 +41,8 @@ public interface DuplicateIDCache { boolean deleteFromCache(byte[] duplicateID) throws Exception; + boolean deleteFromCache(byte[] duplicateID, Transaction tx) throws Exception; + void load(List> ids) throws Exception; void load(Transaction tx, byte[] duplID); 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 c35e4618001..8f012280c09 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 @@ -81,6 +81,11 @@ public boolean deleteFromCache(byte[] duplicateID) { return deleteFromCache(new ByteArray(duplicateID)); } + @Override + public boolean deleteFromCache(byte[] duplicateID, Transaction tx) { + return deleteFromCache(new ByteArray(duplicateID)); + } + private boolean deleteFromCache(final ByteArray duplicateID) { if (logger.isTraceEnabled()) { logger.trace("deleting id = {}", describeID(duplicateID.bytes)); 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 9190487e383..583648ce2b2 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 @@ -55,6 +55,11 @@ public boolean deleteFromCache(byte[] duplicateID) throws Exception { return true; } + @Override + public boolean deleteFromCache(byte[] duplicateID, Transaction tx) throws Exception { + return true; + } + @Override public void load(List> ids) throws Exception { 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 acb202395e3..4d590fc62ff 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 @@ -146,10 +146,16 @@ public synchronized void load(final List> ids) throws Excepti @Override public boolean deleteFromCache(byte[] duplicateID) throws Exception { - return deleteFromCache(new ByteArray(duplicateID)); + return deleteFromCache(new ByteArray(duplicateID), null); } - private boolean deleteFromCache(final ByteArray duplicateID) throws Exception { + @Override + public boolean deleteFromCache(byte[] duplicateID, Transaction tx) throws Exception { + Objects.requireNonNull(tx, "tx must be not null"); + return deleteFromCache(new ByteArray(duplicateID), tx); + } + + private boolean deleteFromCache(final ByteArray duplicateID, final Transaction tx) throws Exception { if (logger.isTraceEnabled()) { logger.trace("deleting id = {}", describeID(duplicateID.bytes)); } @@ -167,7 +173,12 @@ private boolean deleteFromCache(final ByteArray duplicateID) throws Exception { if (logger.isTraceEnabled()) { logger.trace("address = {} deleting id = {}", address, describeID(duplicateID.bytes, id.getB())); } - storageManager.deleteDuplicateID(recordID); + if (tx == null) { + storageManager.deleteDuplicateID(recordID); + } else { + storageManager.deleteDuplicateIDTransactional(tx.getID(), recordID); + tx.setContainsPersistent(); + } } } return true; @@ -393,7 +404,7 @@ public void afterCommit(final Transaction tx) { @Override public void beforeRollback(Transaction tx) throws Exception { if (!afterCommit) { - deleteFromCache(holder); + deleteFromCache(holder, null); } }