fix(binding-mcp-kafka): stop reset_offsets hanging after a stale stage END - #2540
Merged
Conversation
… reuse Closes a gap toward diagnosing #2532 (mcp-kafka reset_offsets hangs against a Dead consumer group): binding-kafka had no coverage for KafkaClientOffsetCommitFactory's bare admin commit (generationId=-1, memberId="", instanceId="") or for chaining multiple apiRequest-kind streams (FindCoordinator, DescribeGroups) on one reused connection, which is exactly how binding-mcp-kafka's reset_offsets flow drives binding-kafka. Add ClientOffsetCommitIT scenarios exercising these paths directly against a live engine: - shouldCommitBareOffset: bare admin commit in isolation - shouldFindCoordinatorOnly / shouldChainFindCoordinatorThenDescribeGroups(NoWait): chained apiRequest-kind streams sharing one affinity/connection, both waiting for and not waiting for the prior stream's END before reopening - shouldResetOffsetsForDeadGroup: the full FindCoordinator -> DescribeGroups (Dead) -> bare OffsetCommit sequence end to end All pass, which rules out KafkaClientApiFactory/KafkaClientOffsetCommitFactory's state machines as the cause of the reported hang in the plaintext, no-SASL case. Root cause investigation continues upstream in binding-mcp-kafka. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Kk7te5cPXiBGfwnVH3cMiq
…e END Fixes #2532. Reproduced against a real Kafka broker (docker compose) with zilla dump/logs and an instrumented build: reset_offsets against a Dead (or Empty) consumer group correctly runs FindCoordinator then DescribeGroups, but the actual bare OffsetCommit result never reaches the MCP client - the broker commits successfully, yet the caller hangs until the MCP session's own inactivity timeout closes the connection. KafkaApiResetOffsetsClient.advanceToOffsetCommit() ends the just-finished DescribeGroups stream and opens the OffsetCommit stream synchronously, setting stage = STAGE_OFFSET_COMMIT before returning. The engine delivers the DescribeGroups stream's own reply-side END acknowledgment on a later turn, by which point onKafkaEnd() sees stage == STAGE_OFFSET_COMMIT and misattributes that stale END as the OffsetCommit stream's own completion, calling peer.doMcpEnd() and closing the MCP response before the real result is ever produced (the subsequent OffsetCommit request and its successful broker response are then orphaned). KafkaApiDescribeConsumerGroupLagClient.onKafkaEnd() already guards against exactly this hazard (comparing end.streamId() against the current kafkaReplyId) for its own two-stage OffsetFetch -> ListOffsets sequence; apply the same guard here. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Kk7te5cPXiBGfwnVH3cMiq
…_offsets against a dead group Adds k3po scripts and IT coverage for the mcp-kafka -> kafka offset-commit handoff (find-coordinator -> describe-groups -> offset-commit) against a consumer group that has never had members, exercising the full 3-stage reset_offsets flow through a live engine (McpKafkaClientIT) as well as a peer-to-peer self-consistency check (KafkaIT), closing the gap that let the stale-END misattribution bug in KafkaApiResetOffsetsClient.onKafkaEnd ship undetected. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Kk7te5cPXiBGfwnVH3cMiq
This was referenced Sep 4, 2026
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
kafka__reset_offsetshanging forever against a consumer group whose state isDead(e.g. one that has never had any members).Root cause:
KafkaApiResetOffsetsClientruns a 3-stage flow (STAGE_FIND_COORDINATOR→STAGE_DESCRIBE_GROUPS→STAGE_OFFSET_COMMIT), reusing the samestagefield and reassigningkafkaInitialId/kafkaReplyIdon each transition.advanceToOffsetCommit()synchronously ends theDescribeGroupsstream and reassignsstage = STAGE_OFFSET_COMMITbefore returning, but the engine delivers that ended stream's own reply-sideENDon a later turn. By the time it arrives,onKafkaEnd()'s guard (stage == STAGE_OFFSET_COMMIT) is already true, so the staleENDfrom the previous stage is misattributed to the newly-openedOffsetCommitstream, and the flow gets stuck without ever sending the offset-commit request or surfacing an error.This was root-caused by reproducing the hang against a real Kafka broker via
docker compose, withzilla dump/instrumented debug logging pinpointing the exact call sequence — not by adding a timeout as a band-aid; the connection was never actually hanging at the transport layer, only misinterpreting an internal event ordering. A sibling class (KafkaApiDescribeConsumerGroupLagClient.onKafkaEnd()) already guards correctly against this by checkingend.streamId() == kafkaReplyIdbefore honoringstage; the same guard is applied here.Fix:
onKafkaEnd()now only advances state / signals MCP completion when theEND'sstreamIdmatches the currentkafkaReplyId, so a staleENDfrom a previous stage transition can no longer be misattributed to the current stage.Test coverage added:
runtime/binding-kafka: newClientOffsetCommitITscenarios covering a bare offset-commit and the find-coordinator → describe-groups chained api-request reuse pattern, backed by new pairedclient.rpt/server.rptk3po scripts at both the application and network layers.runtime/binding-mcp-kafka/specs/binding-mcp-kafka.spec: new end-to-endreset_offsetsscenario against aDeadconsumer group, exercised both peer-to-peer (KafkaIT, self-consistency check) and against a live engine (McpKafkaClientIT#shouldResetOffsetsForDeadGroup) — this is the regression test that reproduces the original hang and confirms the fix.All existing and new tests pass:
binding-mcp-kafka(51/51 ITs, 80/80 unit tests),binding-mcp-kafka.spec(80/80 ITs),binding-kafka+binding-kafka.spec(412/412 tests). Checkstyle and license checks are clean.Fixes #2532
🤖 Generated with Claude Code
https://claude.ai/code/session_01Kk7te5cPXiBGfwnVH3cMiq
Generated by Claude Code