fix(binding-mqtt, binding-mqtt-kafka, binding-kafka): flow control and null-safety defects on the MQTT publish path - #2523
Conversation
|
@sfr-oc please provide the reproduction recipe for the externally observed issue, including the |
Reproduction recipeThe shipped 1. Topic setupIn - kafka-topics.sh --bootstrap-server $${KAFKA_BOOTSTRAP_SERVER} --create --if-not-exists --topic mqtt-devices --config cleanup.policy=compact
+ kafka-topics.sh --bootstrap-server $${KAFKA_BOOTSTRAP_SERVER} --create --if-not-exists --topic mqtt-devices --partitions 5 --config cleanup.policy=compactEverything else in the example is used as-is. The relevant part of north_mqtt_kafka_mapping:
type: mqtt-kafka
kind: proxy
options:
topics:
sessions: mqtt-sessions
messages: mqtt-messages
retained: mqtt-retained
clients:
- place/{identity}/#
routes:
- when:
- publish:
- topic: place/+/device/#
- topic: device/#
- subscribe:
- topic: place/+/device/#
- topic: device/#
with:
messages: mqtt-devices
exit: north_kafka_cache_client
exit: north_kafka_cache_clientThe binding chain is the example's: cd examples/mqtt.kafka.proxy
docker compose up -d2. ClientA single MQTT 3.1.1 client publishing QoS 0 messages, no retain flag. All messages carry the same topic, so they hash to one partition and the remaining four stay idle — that is what makes the two reductions in docker compose exec mosquitto-cli sh -c '
i=0
while [ $i -lt 600 ]; do
echo "{\"seq\":$i,\"test\":\"payload\"}"
sleep 0.01
i=$((i+1))
done | mosquitto_pub -h zilla -p 7183 -d -l \
-t "device/loadtest/state" -q 0 -i "loadtest-client"'Payload is ~30 bytes, which works out to ~542 bytes reserved per message on the produce stream. 3. Observed
60 × 542 ≈ 32768, i.e. The collapse can be watched directly with 4. Without Docker
stalling after the 8th of eleven 1024-byte messages — 8 × 1024 = 8192, the idle partition's maximum, which 5. Originally reported environmentFor completeness, the environment where this first surfaced additionally used TLS with mutual auth on both the MQTT listener and the Kafka client, and a three-broker cluster with a five-partition topic (replication factor 1, no under-replicated partitions). Neither TLS nor the broker count is required to reproduce — the partition count is. |
…tch merged kind KafkaMergedFlushEx is a union discriminated by kind (produce, fetch, consumer). KafkaSignalStream.onKafkaFlush called .fetch() on it unconditionally, without checking that the flush was actually of the fetch sub-kind. Since this stream both produces will/expiry signals and fetches other instances' signals on the same merged Kafka subscription, a produce-kind (or consumer-kind) flush is a normal, frequent occurrence -- every produce ack emits one. Reading .fetch() off a non-fetch-kind union member returns an unwrapped flyweight, and iterating its "progress" array throws a NullPointerException on Array32FW.buffer(), terminating the whole engine worker. Guard the read with the same kind() check already used one line above for the outer union, mirroring the existing pattern in this method. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…indow is exhausted MqttPublishStream.initialBudget() returns initialMax - (initialSeq - initialAck), unlike MqttSessionStream's version which also subtracts initialPad. decodePublishPayload computed the reserved size as Math.max(publisher.initialPad, Math.min(lengthMax + initialPad, initialBudget)), which floors at initialPad even once the window can no longer hold it. Once the granted window is exhausted, this kept emitting zero-length DATA frames that still reserved initialPad bytes, advancing initialSeq with no decode progress. Every subsequent WINDOW re-entered the same path and burned another initialPad, walking initialSeq past initialAck + initialMax. The downstream peer eventually sees sequence > acknowledge + maximum and RESETs, which propagates to onDecodeError with a reason code the MQTT v3.1.1 CONNACK encoding can't carry (reasonCode > MAX_CONNACK_REASONCODE_V4), so doNetworkEnd tears down the connection with no message sent to the client at all - the client just sees the TCP/TLS connection disappear. The threshold is bytes, not messages: it trips as soon as cumulative payload + padding first exhausts the granted initial window, so message count before the cutoff scales inversely with payload size - independent of any will-message handling. Compute the payload cap as Math.min(lengthMax, initialBudget - initialPad) and only proceed once it and the resulting claimed size are non-negative, so once the window is exhausted no frame is emitted and no forward progress is recorded - the stream just waits for the next WINDOW like every other backpressured decode path in this file. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
onClientInitialFlush advanced initialAck itself before asking doClientInitialWindow to publish it. That method only writes a WINDOW frame when it observes the acknowledge or the maximum change, and it derives the new acknowledge from the caller's noAck: with initialAck already advanced, initialSeq - noAck resolves back to the current initialAck, so the guard never fires and the frame is never written. The credit is released internally and never reaches the sender, whose view of the window shrinks by every flush until it reaches zero. Pass the outstanding byte count without pre-applying it, so doClientInitialWindow advances the acknowledge and emits the frame, and keep the maximum from regressing below the one already advertised. With a zero-reserved flush this resolves to the previous behaviour. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…g the publish window A publish proxy on a route that has a retained topic configured charges the retained stream a full reserved on every non-retained PUBLISH, because onMqttData issues a flush on it for each message that does not carry the retain flag. The window granted upstream, however, was derived from the messages stream alone: doMqttWindow guarded the retained-aware computation with hasPublishFlagRetained, a per-message flag that onMqttData resets after every FIN, leaving the branch unreachable in steady state. The retained stream's sequence could therefore advance past the window it was granted, and its credit never constrained what the client was allowed to send. Gate the computation on retainAvailable, the per-stream capability that actually decides whether the retained stream is charged. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
… publish window The messages and retained streams advertise credit in different shapes: the messages stream holds its acknowledge at zero and grows its maximum, while the retained stream advances its acknowledge against a fixed maximum. Minimizing acknowledge and maximum independently across the two therefore pairs the acknowledge of one with the maximum of the other, describing neither stream's actual budget, and caps the window offered upstream at a fixed maximum whose acknowledge never moves - a limit on the total bytes a connection can publish rather than on bytes in flight. Compare the budget each stream offers and express the smaller of the two against the chosen maximum, keeping the acknowledge monotonic. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…udget offered doMergedInitialWindow aggregated the produce partitions by taking the largest unacknowledged byte count across them and, separately, the smallest maximum. A partition carrying traffic holds its acknowledge and grows its maximum, while idle partitions keep the maximum they were opened with, so those two reductions select different partitions: the unacknowledged bytes of the busiest one get paired with the maximum of an idle one. The merged window then shrinks by every byte written until it reaches zero and the stream stalls for good. With a single partition both reductions select the same stream, which is why this only appears on topics with more than one partition. Reduce over the budget each partition actually offers instead, and add the unacknowledged bytes back to express it as a maximum. For a single partition this is arithmetically identical to the previous computation. Adds a merged window trace under the existing produce debug property, since this aggregation was not observable from a running gateway. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…tition merged.produce.message.values.partition.idle opens two produce partitions, sends all traffic to one and leaves the other idle. Reducing the maximum across partitions independently of the unacknowledged bytes pins the merged window to the idle partition's maximum while the busy partition's unacknowledged bytes keep growing, so the stream stalls once that maximum is reached; the scenario asserts a further write still succeeds beyond it. Co-Authored-By: Claude Opus 5 (1M context) <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. Reachable independently of the QoS2+Will+CleanStart timing race already covered by shouldSendWillSignalOnQos2AbortBeforeSessionEstablished: any uncaught exception on the engine worker triggers EngineWorker.onClose's forced synthetic-abort teardown, which calls onMqttAbort -> sendWillSignal / sendExpirySignal -> doKafkaData on every still-open MqttSessionProxy, including ones whose Kafka-side session stream never finished attaching. That unconditional receiver.accept(...) in the shared doData helper then NPEs, escalating the original fault into a second crash instead of a clean per-worker restart. Add the same kafka != null guard to all three doKafkaData overloads, at the single shared point that covers sendWillSignal, sendExpirySignal, and every other caller. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
c027cf6 to
8c97f33
Compare
|
@sfr-oc thank you for filing the issues and providing most of the fixes in your PR. 🙏 We've split this PR into a separate PR per issue, each with test coverage and following IT scenario naming conventions. The reproduction recipe now works as desired. The fixes will be included in the next 2.x release, which is imminent. |
2 participants
Summary
An MQTT client publishing small QoS 0 messages through an
mqtt-kafkaproxy to a multi-partition Kafka topic stops being served after roughly 60 messages. Zilla closes the connection itself, with no exception and no event in the log — the client only observes that the connection disappeared. Chasing that symptom surfaced seven distinct defects across three bindings; they are reported individually and fixed here in one commit each, so any subset can be taken independently.One of the seven, #2522, has meanwhile been fixed upstream in #2534 and is no longer part of this branch, which has been rebased onto current
develop.The reported message count varies with payload size and socket buffering, which is why the same defect was observed at 42, 61 and 239 messages before its cause was understood.
Commits
doKafkaDataagainst unattached kafka streambinding-mqtt-kafkaKafkaSignalStreamflush against non-fetch merged kindbinding-mqtt-kafkabinding-mqttbinding-kafkabinding-mqtt-kafkabinding-kafkaThe primary defect (#2516)
KafkaMergedFactory.doMergedInitialWindowaggregates produce partitions with two independent reductions — the largest unacknowledged byte count, and the smallest maximum. A partition carrying traffic holds its acknowledge and grows its maximum, while idle partitions keep the maximum they were opened with, so the two reductions select different partitions and pair values that describe neither one's budget. The merged window then shrinks by every byte written until it reaches zero. With a single partition both reductions select the same stream, which is why this only appears on topics with more than one partition.Verification
merged.produce.message.values.partition.idleopens two produce partitions, sends all traffic to one and leaves the other idle. Without the fix it stalls at the idle partition's maximum:Also reproduced end to end against a live engine (
examples/mqtt.kafka.proxy, Kafka 4.1.1, 5-partition topic):Regression across the touched modules and their spec projects —
binding-kafka,binding-mqtt-kafka,binding-mqttand each.spec— passes with 0 failures, checkstyle and license checks clean.An earlier revision of #2516 broke
CacheMergedIT/ClientMergedITshouldProduceMergedMessageValue100k; the committed form is arithmetically identical to the previous computation for a single partition, and both tests pass.Test coverage — please read
Coverage is uneven and I would rather say so up front:
reservedimmediately and only ever writesreserved=0flushes, so the condition cannot be constructedStatus
Verified against the reporting environment's Kafka test cluster: 1000 messages published without a disconnect, where the same client previously stalled. Not yet verified in that environment's production deployment.
Given how much of this was found by inspection rather than by a failing test, independent verification before merging would be welcome — in particular #2520 and #2518, where the remedy involves a judgement call rather than a mechanical correction. #2520 guards a write that would otherwise throw; if the intended behaviour is to defer it until the stream attaches, or to fail the session explicitly, the patch should follow that instead.