fix(binding-mqtt, binding-mqtt-kafka): prevent engine crash on MQTT CONNECT with Will message - #2429
Open
sfr-oc wants to merge 2 commits into
Open
fix(binding-mqtt, binding-mqtt-kafka): prevent engine crash on MQTT CONNECT with Will message#2429sfr-oc wants to merge 2 commits into
sfr-oc wants to merge 2 commits into
Conversation
…nt session window When a client sends CONNECT with a Will message, onDecodeConnectWillPayload computed payloadSize as session.initialBudget() - headerSize with no floor. The guarding MqttState.initialOpened(session.state) check only proves a WINDOW frame arrived at least once for the session stream, not that the granted window is large enough to hold the encoded will-message header. When it isn't, the subtraction goes negative, producing a negative OctetsFW.wrap() limit and an uncaught IndexOutOfBoundsException that terminates the engine worker. Add a session.hasSessionWindow(headerSize) check before building the header, mirroring the same pattern already used by the SUBSCRIBE-to-session-state sync path. When the window is too small, return progress unchanged so the existing decodeSlot buffering parks the bytes and decodeNetwork retries automatically on the next session WINDOW. Fixes aklivity#2428 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…tream KafkaSessionStream.kafka is only assigned inside a subclass's doKafkaBegin override, so it stays null until that handshake step runs. Every other outbound method on the class (doKafkaEnd, doKafkaAbort, doKafkaReset, doKafkaWindow) already guards on kafka != null, but the three doKafkaData overloads did not, so any caller reaching them first (e.g. sendWillSignal via onMqttAbort) hit an NPE in the shared doData helper's unconditional receiver.accept(...). This is reachable in normal operation: a client that negotiates publishQosMax == 2 with Will + Clean Start, then aborts before doCreateSessionStream has run, leaves the session's kafka field unset when onMqttAbort fires sendWillSignal. Add the same kafka != null guard to all three doKafkaData overloads, at the single shared point that covers sendWillSignal and every other caller. Fixes aklivity#2428 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
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
1 participant
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.
Fixes #2428
Two independent bugs combined to let a single MQTT client with a Last Will and Testament crash the whole engine (and in production, segfault the JVM):
onDecodeConnectWillPayloadcomputed a payload size fromsession.initialBudget() - headerSizewith no floor. When the session stream's granted window is smaller than the encoded will-message header alone, this goes negative and corrupts a subsequentOctetsFW.wrap()call, throwing an uncaughtIndexOutOfBoundsExceptionthat terminates the engine worker. Fixed by checkingsession.hasSessionWindow(headerSize)first, mirroring the existing SUBSCRIBE-to-session-state-sync pattern in the same file.KafkaSessionStream.doKafkaDatahad nokafka != nullguard, unlike every sibling method (doKafkaEnd/doKafkaAbort/doKafkaReset/doKafkaWindow). Reachable independently of bug 1 whenever a client negotiates QoS 2 with Will + Clean Start and aborts before the Kafka-side session stream is established —sendWillSignalthen NPEs on the unattached stream. This also amplified bug 1's crash into a full JVM crash, since it fired again during the engine's forced teardown.Both fixes are minimal and follow existing idiomatic patterns in their respective files. Each has a new failing-first k3po spec scenario (confirmed to reproduce the exact production stack traces pre-fix) plus full regression coverage on both modules — see the two commits for details.
Test plan
session.will.message.header.exceeds.window) fails pre-fix with the reportedIndexOutOfBoundsException, passes post-fixsession.will.message.qos2.abort.before.session.established) fails pre-fix with the reported NPE, passes post-fixruntime/binding-mqtt,runtime/binding-mqtt-kafka,specs/binding-mqtt.spec,specs/binding-mqtt-kafka.speccheckstyle:checkclean on all four modules