fix(binding-mqtt-kafka): guard doKafkaData against insufficient window budget

[jfallows]

Description

KafkaSessionStream.doKafkaData (three overloads, in MqttKafkaSessionFactory) called doData(kafka, ...) unconditionally, then asserted initialSeq - padding <= initialAck + initialMax after the fact. KafkaSessionStateProxy's own kafka producer stream (topic mqtt-sessions) gets its kafka field assigned synchronously as soon as doKafkaBeginIfNecessary runs, but initialMax stays at its Java default 0 until the merged Kafka binding's async partition/metadata DESCRIBE resolves and grants a real window (see KafkaMergedFactory.onMergedInitialBegin, which kicks off describeStream.doDescribeInitialBegin rather than synchronously granting credit). A client that aborts with its Will flag set during that gap reaches onMqttAbort → sendWillSignal → doKafkaData with a real (non-null) kafka stream but zero window, tripping the assert and crashing the entire engine worker via AgentTerminationException — every connection on that worker goes down with it.

The original community-reported fix (credit to @sfr-oc for identifying and reporting this in #2520) guarded only the narrower kafka == null case (reachable via EngineWorker.onClose's synthetic-abort sweep hitting a stream mid-registration). This PR instead guards each doKafkaData overload with the same inequality the existing assert already expressed, checked before attempting the write instead of after. This subsumes the kafka == null case for free (an unattached stream always has initialMax == 0) while also covering the more general, more reachable "attached but zero window" case that the narrower guard didn't touch.

New coverage: session.will.message.clean.start.abort.zero.window grants the session's kafka stream a real window just large enough for the self-covering expiry-signal writes but too small for the will signal, then aborts. Verified red (reverting the fix reproduces the exact AssertionError/AgentTerminationException crash, killing engine/worker#0) then green, plus MqttIT/KafkaIT peer-to-peer self-consistency coverage. Full runtime/binding-mqtt-kafka and specs/binding-mqtt-kafka.spec module suites pass.

Fixes #2520

🤖 Generated with Claude Code

https://claude.ai/code/session_015YVNaqKvEXZVzmg3HoGVnt

---

Generated by Claude Code