ARTEMIS-6189 QoS 1 & 2 message handling is not resilient - #6613
Open
jbertram wants to merge 4 commits into
Open
Conversation
The JournalHashMap collection ID was hardcoded as a primitive long. This generalizes it to a type parameter, enabling journal hash maps keyed by arbitrary types such as String. This is needed for MQTT packet-ID correlation, which is keyed by client ID. - Collection ID encoding/decoding is now delegated to the persister via abstract methods, rather than hardcoded as long read/write - The internal map storage switches from Netty's LongObjectHashMap to a standard HashMap to support non-primitive keys - Null-safety fix in JournalHashMap.remove() prevents an NPE when removing a key that doesn't exist Co-Authored-By: Claude <noreply@anthropic.com>
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 <noreply@anthropic.com>
Add comprehensive tests for MQTT QoS 1 and QoS 2 resiliency across broker restarts and client reconnects. These tests use interceptors to interrupt message flows at every stage of the QoS handshake and verify correct redelivery, exactly-once semantics, and proper state cleanup. - QoS 1 subscriber tests cover broker restart and client disconnect both before and after PUBACK, verifying packet ID reuse with DUP flag - QoS 1 publisher tests are not needed because the broker has no intermediate state to maintain; it either persists the inbound message or it doesn't, and the client drives retransmission entirely on its own - QoS 2 publisher tests cover interruptions at all four stages of the handshake (before/after PUBREC, before/after PUBCOMP) for both broker restart and client disconnect - QoS 2 subscriber tests cover the same stages plus session expiry cleanup via zero-interval disconnect, timed scanner, and clean-start reconnect - A soak test runs many publishers and many subscribers sending many QoS 2 messages with periodic broker restarts, verifying exactly-once delivery - HiveMQ MQTT 5 client added as a test dependency Co-Authored-By: Claude <noreply@anthropic.com>
During repeated execution of MQTT tests I fixed these flaky tests: - Add a notification listener to the cluster remote subscribe test to synchronize on SESSION_CREATED, preventing a race in subscription takeover - Fix non-deterministic message ordering in testAutoDeleteRetainedQueue by consuming the retained message before publishing non-retained messages - Use Wait.assertFalse for async disconnect detection instead of an immediate assertion Co-Authored-By: Claude <noreply@anthropic.com>
| * @return An array of integers representing the list of accepted QoS for each topic | ||
| */ | ||
| int[] addSubscriptions(List<MqttTopicSubscription> subscriptions, Integer subscriptionIdentifier) throws Exception { | ||
| session.awaitStart(); |
Contributor
There was a problem hiding this comment.
This kind of thing won't scale very well... you will have a thread holding the start. If you have many threads starting you may starve and never be able to complete.
Isn't possible to fully play with callbacks?
| IllegalStateException unableToGenerateID(); | ||
|
|
||
| @Message(id = 850002, value = "StorageManager is null") | ||
| IllegalStateException storageManagerIsNull(); |
Contributor
There was a problem hiding this comment.
I wouldn't use the Bundle for such thing.. StorageManager null is clearly an internal error, not supposed to happen. I would probably just use assert storageManager != null in its palce.
| private Queue sessionStore; | ||
| private final Map<String, MQTTConnection> connectedClients = new ConcurrentHashMap<>(); | ||
| private final boolean subscriptionPersistenceEnabled; | ||
| private final JournalHashMapProvider<String, PacketIdCorrelationKey, Integer, Object> journalHashMapProvider; |
Contributor
There was a problem hiding this comment.
what's the lifecycle of these records? is there any cleanup in place? how many records are we going to create here? one per session? one per ack? I'm kind of lost here about
| public PacketIdCache(MQTTSession session, TYPE type) { | ||
| this.session = session; | ||
| this.postOffice = session.getServer().getPostOffice(); | ||
| this.cacheName = getCacheName(session.getServer().getInternalNamingPrefix(), session.getState().getClientId(), type); |
Contributor
There was a problem hiding this comment.
from what I understand you're creating a duplicate_ID_cache based on the clientID.
What's the cleanup. Say I connect now with a clientID.. and never again?
I can create enough records to OME the server? or is there a cleanup?
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This PR implements MQTT QoS 1 and QoS 2 resiliency so that in-flight message flows survive broker restarts/crashes and client reconnects.
See the Jira for details about exactly what isn't working.
See the individual commit messages for an overview of what's been changed.
I know this is big, but this was a complex task that required refactoring many core pieces of the MQTT implementation. It expanded a lot as I was working through it. Claude helped along the way.
The full test-suite is being run now, but I wanted to get it out for review sooner than later.