[fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close - #26433
Open
merlimat wants to merge 2 commits into
Open
[fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close#26433merlimat wants to merge 2 commits into
merlimat wants to merge 2 commits into
Conversation
A scalable consumer's registration lives on the controller, but its connection comes from the shared connection pool: a consumer's close never makes the channel inactive while the client process lives, so the controller never noticed a clean departure — the registration stayed "connected" indefinitely (a same-process leave was a permanent zombie group member) and even a cross-process clean leave stalled the rebalance for the full disconnect grace period. Adds CommandScalableTopicUnsubscribe (BaseCommand type 82): on close the client tells the controller to delete the registration and rebalance the group immediately; the disconnect grace period remains the fallback for unclean departures. - Client: the unsubscribe is chained on the in-flight subscribe attempt's outcome, so it can never overtake the registration (the broker records the per-connection registration ref before sending the subscribe response); best-effort — a failure never fails the close. - Broker: the handler resolves the consumer through the connection's own registration map (so a client can only unregister sessions it created), forwards to the existing SubscriptionCoordinator.unregisterConsumer (grace-timer cancel + persisted-entry delete + rebalance), and answers CommandSuccess, idempotently for unknown ids. The registration ref is removed only after the unregister succeeds, keeping the channelInactive grace fallback alive if the explicit path fails. The rejoin e2e now runs against the default grace period with all consumers on one shared client: the leave hands the segment back within seconds purely through the unsubscribe path (previously it required a dedicated client per leaver and a shrunken grace period). Assisted-by: Claude Code (Fable 5)
…sumer-unsubscribe
lhotari
dismissed stale reviews from themself
August 29, 2026 20:09
Retracted: posted in error by a local tooling test.
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
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.
Motivation
Follow-up promised in #26208: a scalable consumer's registration lives on the controller, but the controller connection comes from the shared connection pool — so a consumer's clean close never makes a channel inactive while the client process lives, and the controller never noticed the departure. A same-process leave (several consumers in one process is the common case) left a zombie group member indefinitely: the disconnect grace timer never even started, the group never rebalanced, and the leaver's segment was never handed back. Even a cross-process clean leave stalled the rebalance for the full grace period (default 60s).
Modifications
Wire format
CommandScalableTopicUnsubscribe { request_id, consumer_id }(BaseCommandtype 82): a scalable consumer cleanly leaving its subscription. Acknowledged with the standardCommandSuccess/CommandError; idempotent — an unknownconsumer_id(e.g. already swept by a disconnect) still succeeds. No new response message.Broker
ServerCnx.handleCommandScalableTopicUnsubscribe: resolves the consumer through the connection's own registration map, so a client can only unregister sessions it created on that connection — no separate authorization needed. Forwards to the existingSubscriptionCoordinator.unregisterConsumer(cancels the grace timer, deletes the persisted registration, rebalances and notifies the remaining consumers) via a newScalableTopicServicepassthrough.channelInactivesweep can still report the disconnect later, so the grace-period fallback stays alive.Client
ScalableConsumerClient.close()sends the unsubscribe, chained on the in-flight subscribe attempt's outcome: the broker records the registration ref before sending the subscribe response, so the unsubscribe can never overtake its own registration (an early unsubscribe would no-op and leave a ghost behind). Best-effort throughout — a failure never fails the close; the grace period remains the fallback for unclean departures.The disconnect/grace machinery is unchanged and still covers crashes and connection loss.
Verifications
testConsumerRejoiningAfterLeaveDoesNotWedgeReleasenow runs with all consumers on one shared client against the default 60s grace period: the leave hands the whole segment back within seconds purely through the unsubscribe path (previously the test needed a dedicated client per leaver plus a shrunken grace period to see a departure at all).