fix(binding-kafka): reconnect KIND_API_REQUEST pool after idle teardown - #2544
Merged
Conversation
KafkaApiClient pools a single network connection per affinity across many independent Kafka API requests. When that connection was reset or aborted while idle (no request in flight, none queued), cleanup only OR'd the CLOSED bits into state without clearing the sticky OPENING/OPENED bits, so the next unrelated request's enqueue() saw a connection that looked both open and closed, took the "still open" branch, and wrote its request onto the dead stream. No BEGIN was ever resent and no response ever arrived, hanging the caller indefinitely. Each of onNetEnd/onNetAbort/onNetReset, and the internal cleanupNet/ cleanupNetPending teardown paths, now converge on a single onNetClosed() once both directions are confirmed closed: it fails the app-side backlog, releases decode/encode slots and budget, resets the connection's bookkeeping to a clean slate, and schedules the reconnect — restoring the invariant that a closed connection's state never lingers as "open" for a later caller to trip over. Fixes #2532 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XJZqfbhycCM4bTy2u2Z7qc
jfallows
force-pushed
the
claude/zilla-issue-2532-t590ti
branch
from
September 4, 2026 17:06
9e8eb5b to
942e931
Compare
doNetEnd/doNetAbort/doNetReset are always the last mutation for whichever direction they close, and any preceding direct state assignment in a caller already happened before they run — so checking KafkaState.closed(state) at the end of each of them is safe for every caller, not just the three onNetXxx handlers that already check explicitly. This covers cleanupNet() and cleanupNetPending() (the internal decode-error/SASL-failure teardown paths), which previously needed their own trailing check to get the same guarantee and no longer do. The onNetXxx handlers keep their own explicit check too, so onNetClosed() is evaluated twice on that path (once inside doNetEnd/Abort/Reset, once in the caller) — harmless, since the first evaluation resets state to 0, making the second trivially false. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XJZqfbhycCM4bTy2u2Z7qc
…reset-reconnect The new create.topics.v7.idle.reset.reconnect scenario had a client.rpt under streams/application/api/ but no matching server.rpt or peer-to-peer IT method, leaving the pairing/self-consistency check specs/AGENTS.md requires for every scenario unmet (an existing sibling scenario, create.topics.v7.reconnect.no.probe, carries the same pre-existing gap, left alone here as out of scope). The app-layer server.rpt also emits the CONNECTION_RESET notify the shared client.rpt awaits before its second connect, since notify/await barriers are session-wide: in the runtime IT that signal comes from the network script after the simulated idle-connection reset, but the pure app-layer peer test has no network side, so the server script provides it instead to unblock the second exchange. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XJZqfbhycCM4bTy2u2Z7qc
…ng apiRequest BEGIN KafkaApiStream.onAppWindow marked the reply as both opening and opened via KafkaState.openedReply(state) whenever it received a reply-window grant, on the assumption a window could only arrive after this stream's own doAppBegin had already sent BEGIN. A proactive reply-window grant (originating from the http server binding's early-window optimization and propagating down through mcp/mcp-kafka/kafka) can now arrive before the real response does, so REPLY_OPENING was getting set before doAppBegin ever ran. doAppBegin's `if (!KafkaState.replyOpening(state))` guard then saw the bit already set and silently skipped sending BEGIN, while the response DATA/END still went out - leaving reset_offsets (and any other multi-stage apiRequest chain, e.g. FindCoordinator -> DescribeGroups) hung forever waiting for a BEGIN that would never come. Adds KafkaState.openReply(state), setting REPLY_OPENED alone, matching the independent opening/opened bit pattern HttpState already uses (see HttpClientFactory/HttpServerFactory). onAppWindow now uses openReply instead of the combined openedReply, so a window grant only ever records flow-control progress and never implies BEGIN was sent - doAppBegin remains the sole setter of REPLY_OPENING. Also adds a network-level regression test exercising the FindCoordinator -> DescribeGroups chain with api.versions negotiation enabled (the production default) rather than disabled, since every existing test in this class ran with api.versions off and so never exercised this pooled-connection request-dispatch path. Fixes #2532 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
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.
Description
Fixes the deeper, more fundamental hang described in this comment on #2532, re-traced after #2540 was merged.
Root cause:
KafkaClientApiFactory.KafkaApiClient(backing everyKIND_API_REQUESTclient, includingbinding-mcp-kafka'sKafkaApiResetOffsetsClientand friends) pools a single network connection peraffinityacross many independent Kafka API requests over time. When that connection is reset or aborted while idle (no request in flight, none queued — e.g. the south TCP connection silently drops after being idle), cleanup only OR'd theINITIAL_CLOSED/REPLY_CLOSEDbits intostate, without ever clearing the stickyINITIAL_OPENING/REPLY_OPENEDbits left over from when the connection was first opened.stateends up simultaneously "opened" and "closed".The next, unrelated caller's
enqueue()only checked the opening/opened bits (neverclosed), so it took the "connection is still usable" branch and wrote its request straight onto the dead stream — no freshBEGINwas ever sent, and no response (success or error) ever arrived, hanging the caller indefinitely. This matches the wire trace in the linked comment exactly: aDATAframe reusing an already-reset stream id with no precedingBEGIN.Fix:
onNetEnd()/onNetAbort()/onNetReset()(and the internalcleanupNet()/cleanupNetPending()teardown paths used by decode-error handling) now each converge on a singleonNetClosed()once both directions are confirmed closed (KafkaState.closed(state)).onNetClosed()fails the app-side backlog, releases the decode/encode slots and budget debitor, resets the connection's bookkeeping (sequence numbers, decoder, correlation id) to a clean slate, and schedules the reconnect — restoring the invariant that a torn-down connection'sstatenever lingers as "open" for a later caller to trip over.doNetBegin()'s own now-redundant reset branch is removed, andenqueue()reverts to its original, simpler form sincestateis trustworthy everywhere again.Test coverage added:
specs/binding-kafka.spec: newcreate.topics.v7.idle.reset.reconnectscenario (application + network scripts, usingnotify/awaitbarriers to force the idle-teardown-then-new-request ordering deterministically), plus a network-layer peer-to-peer self-consistency IT (KafkaCreateTopicsIT#shouldReconnectAfterIdleConnectionReset).runtime/binding-kafka:KafkaCreateTopicsIT#shouldReconnectAfterIdleConnectionReset— reproduces the exact hang against a live engine (confirmed as a failing/timing-out test against the pre-fix code) and confirms the fix.All 47 IT classes in
runtime/binding-kafka(218 unit tests, all integration tests including SASL, cache, fetch/produce, group, and every existing reconnect scenario) pass with no regressions. Checkstyle and license checks are clean on both changed modules.Fixes #2532
🤖 Generated with Claude Code
https://claude.ai/code/session_01XJZqfbhycCM4bTy2u2Z7qc
Generated by Claude Code