Skip to content

[fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close - #26433

Open
merlimat wants to merge 2 commits into
apache:masterfrom
merlimat:mmerli/scalable-consumer-unsubscribe
Open

[fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close#26433
merlimat wants to merge 2 commits into
apache:masterfrom
merlimat:mmerli/scalable-consumer-unsubscribe

Conversation

@merlimat

Copy link
Copy Markdown
Contributor

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 } (BaseCommand type 82): a scalable consumer cleanly leaving its subscription. Acknowledged with the standard CommandSuccess/CommandError; idempotent — an unknown consumer_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 existing SubscriptionCoordinator.unregisterConsumer (cancels the grace timer, deletes the persisted registration, rebalances and notifies the remaining consumers) via a new ScalableTopicService passthrough.
  • The per-connection registration ref is removed only after the unregister succeeds: if the explicit path fails (e.g. a metadata-store error), the channelInactive sweep 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

  • testConsumerRejoiningAfterLeaveDoesNotWedgeRelease now 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).
  • Full sweep: every v5 client suite (the Docker-based multi-broker suite excluded — no Docker locally), the scalable broker suites, and checkstyle across the touched modules, re-run after merging latest master.

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)
@merlimat
merlimat requested a review from lhotari August 28, 2026 22:49
lhotari

This comment was marked as outdated.

lhotari

This comment was marked as outdated.

lhotari

This comment was marked as outdated.

lhotari

This comment was marked as outdated.

lhotari

This comment was marked as outdated.

@lhotari
lhotari dismissed stale reviews from themself August 29, 2026 20:09

Retracted: posted in error by a local tooling test.

@lhotari lhotari added this to the 5.0.0-M2 milestone Aug 31, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants