Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
merlimat wants to merge 2 commits into
base: master
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
dddb4a6
[fix][client] Scalable consumers: explicit unsubscribe on clean close
merlimat Aug 7, 2026
2a1f44d
Merge remote-tracking branch 'apache/master' into mmerli/scalable-con…
merlimat Aug 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 40 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,6 +142,7 @@
import org.apache.pulsar.common.api.proto.CommandScalableTopicClose;
import org.apache.pulsar.common.api.proto.CommandScalableTopicLookup;
import org.apache.pulsar.common.api.proto.CommandScalableTopicSubscribe;
import org.apache.pulsar.common.api.proto.CommandScalableTopicUnsubscribe;
import org.apache.pulsar.common.api.proto.CommandSeek;
import org.apache.pulsar.common.api.proto.CommandSend;
import org.apache.pulsar.common.api.proto.CommandSubscribe;
Expand Down Expand Up @@ -1177,6 +1178,45 @@ protected void handleCommandScalableTopicSubscribe(
});
}

@Override
protected void handleCommandScalableTopicUnsubscribe(
CommandScalableTopicUnsubscribe commandScalableTopicUnsubscribe) {
checkArgument(state == State.Connected);
final long requestId = commandScalableTopicUnsubscribe.getRequestId();
final long consumerId = commandScalableTopicUnsubscribe.getConsumerId();

// The lookup is scoped to this connection's own registrations, so a client can only
// unregister sessions it created here — no further authorization is needed.
ScalableConsumerRegistrationRef ref = scalableConsumerRegistrations.get(consumerId);
var scalableTopicService = service.getScalableTopicService();
if (ref == null || scalableTopicService == null) {
// Unknown or already swept by a disconnect: idempotent success.
getCommandSender().sendSuccessResponse(requestId);
return;
}
log.debug().attr("topic", ref.topicName()).attr("subscription", ref.subscription())
.attr("consumerName", ref.consumerName()).attr("requestId", requestId)
.log("Received ScalableTopicUnsubscribe");
scalableTopicService.unregisterConsumer(ref.topicName(), ref.subscription(), ref.consumerName())
.whenCompleteAsync((__, ex) -> {
if (ex != null) {
// Keep the ref: the channelInactive sweep can still report the
// disconnect, so the grace-period fallback stays alive for a
// registration the explicit unregister failed to delete.

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[BUG] the grace-period fallback this comment promises is not preserved: a failed metadata delete leaves the group wedged with no timer and no rebalance

The comment on these lines states the invariant the whole error path rests on: "Keep the ref: the channelInactive sweep can still report the disconnect, so the grace-period fallback stays alive for a registration the explicit unregister failed to delete." It does not hold.

SubscriptionCoordinator.unregisterConsumer (SubscriptionCoordinator.java:220-236) drops the in-memory session and cancels the grace timer before the async metadata delete:

ConsumerSession removed = sessions.remove(consumerName);
if (removed == null) { return CompletableFuture.completedFuture(snapshotAssignments()); }
removed.cancelGraceTimer();
return resources.unregisterConsumerAsync(topicName, subscriptionName, consumerName)
        .thenApply(__ -> { ... rebalanceAndNotify() ... });

If the delete fails, the future fails and thenApply — the rebalance — never runs. The promised fallback is then a no-op: channelInactive → ScalableTopicService.onConsumerDisconnect → SubscriptionCoordinator.onConsumerDisconnect (:245-250) opens with sessions.get(consumerName); if (session == null || !session.isConnected()) return; and the session is already gone, so no grace timer is ever armed. There is no delete retry and no periodic registration reconciliation; restoreConsumers only runs on controller initialization / leader recovery.

Failure scenario. A metadata-store error (session loss, store unavailable) during an ordinary clean consumer close. Afterwards: the in-memory session is gone, segmentAssignments still points at it because no rebalance ran, the persisted registration survives, the grace timer is cancelled, and nothing retries — the departed consumer's segment is never handed back to the group. That is exactly the wedge this PR exists to remove, now reachable on the common close path (before this PR, the explicit unregister was not on that path).

Two related notes:

  • evictExpiredConsumer (SubscriptionCoordinator.java:~505-531) handles the identical failure better — .exceptionally(… return null).thenRun(… rebalanceAndNotify …) — so it still rebalances. The same operation now has two different failure behaviours, and the newly-common path has the worse one.
  • The advertised idempotency ("an unknown consumer_id still succeeds") makes this stickier rather than safer: once the first delete has failed, the session is absent, so any repeat call returns success immediately at :222-224 without retrying the deletion — and ServerCnx would then drop its retained ref, removing even the nominal fallback.

Moving sessions.remove / cancelGraceTimer after a successful delete in SubscriptionCoordinator.unregisterConsumer, or mirroring the eviction path's .exceptionally(...).thenRun(rebalance), would make the comment true.

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[BUG] the grace-period fallback this comment promises is not preserved: a failed metadata delete leaves the group wedged with no timer and no rebalance

The comment on these lines states the invariant the whole error path rests on: "Keep the ref: the channelInactive sweep can still report the disconnect, so the grace-period fallback stays alive for a registration the explicit unregister failed to delete." It does not hold.

SubscriptionCoordinator.unregisterConsumer (SubscriptionCoordinator.java:220-236) drops the in-memory session and cancels the grace timer before the async metadata delete:

ConsumerSession removed = sessions.remove(consumerName);
if (removed == null) { return CompletableFuture.completedFuture(snapshotAssignments()); }
removed.cancelGraceTimer();
return resources.unregisterConsumerAsync(topicName, subscriptionName, consumerName)
        .thenApply(__ -> { ... rebalanceAndNotify() ... });

If the delete fails, the future fails and thenApply — the rebalance — never runs. The promised fallback is then a no-op: channelInactive → ScalableTopicService.onConsumerDisconnect → SubscriptionCoordinator.onConsumerDisconnect (:245-250) opens with sessions.get(consumerName); if (session == null || !session.isConnected()) return; and the session is already gone, so no grace timer is ever armed. There is no delete retry and no periodic registration reconciliation; restoreConsumers only runs on controller initialization / leader recovery.

Failure scenario. A metadata-store error (session loss, store unavailable) during an ordinary clean consumer close. Afterwards: the in-memory session is gone, segmentAssignments still points at it because no rebalance ran, the persisted registration survives, the grace timer is cancelled, and nothing retries — the departed consumer's segment is never handed back to the group. That is exactly the wedge this PR exists to remove, now reachable on the common close path (before this PR, the explicit unregister was not on that path).

Two related notes:

  • evictExpiredConsumer (SubscriptionCoordinator.java:~505-531) handles the identical failure better — .exceptionally(… return null).thenRun(… rebalanceAndNotify …) — so it still rebalances. The same operation now has two different failure behaviours, and the newly-common path has the worse one.
  • The advertised idempotency ("an unknown consumer_id still succeeds") makes this stickier rather than safer: once the first delete has failed, the session is absent, so any repeat call returns success immediately at :222-224 without retrying the deletion — and ServerCnx would then drop its retained ref, removing even the nominal fallback.

Moving sessions.remove / cancelGraceTimer after a successful delete, or mirroring the eviction path's .exceptionally(...).thenRun(rebalance), would make the comment true.

Throwable cause = ex.getCause() != null ? ex.getCause() : ex;
log.warn().attr("consumerName", ref.consumerName()).exceptionMessage(cause)
.log("ScalableTopicUnsubscribe failed");
getCommandSender().sendErrorResponse(requestId, ServerError.UnknownError,
cause.getMessage());
return;
}
// Removed only on success; a channelInactive racing the unregister just
// re-reports an already-removed session, which the coordinator ignores.
scalableConsumerRegistrations.remove(consumerId, ref);
getCommandSender().sendSuccessResponse(requestId);
}, ctx.executor());
}

@Override
protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata partitionMetadataParam) {
checkArgument(state == State.Connected);
Expand Down
15 changes: 15 additions & 0 deletions ...-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -316,6 +316,21 @@ public void onConsumerDisconnect(TopicName topic, String subscription, String co
}
}

/**
* Explicit clean leave: forwards to the locally-held controller, which deletes the
* persisted registration and rebalances the remaining consumers immediately. No-op when
* the controller is not held locally (e.g. leadership moved) — the disconnect grace

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[QUALITY] the javadoc's 'e.g. leadership moved' names a case that cannot reach this no-op branch

Small, but it misdescribes the fallback contract my ServerCnx comment above is about. The javadoc says the no-op branch covers the case where "the controller is not held locally (e.g. leadership moved)" — and ServerCnx reads the resulting completedFuture(null) as success and removes the per-connection registration ref.

I first read this as a fallback-loss bug, and that reading is wrong; the counter-argument is worth recording: leadership movement does not remove the controllers map entry. Registration installs the future via getOrCreateController (:100-112), a deposed controller stays in the map, and ScalableTopicController.unregisterConsumer's checkLeader() then makes the future complete exceptionally — so ServerCnx takes the error branch and keeps the ref, which is the correct behaviour. Entries are removed only on failed initialization, an explicit releaseController (whose sole production caller is topic deletion), or service shutdown.

So the branch is fine; the example in the javadoc names a case that cannot reach it, which is misleading in a comment whose whole job is to explain when the grace-period fallback still applies. Suggest replacing "e.g. leadership moved" with the cases that actually reach it (topic deleted, service shutting down, or a consumer that never registered here).

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[QUALITY] the javadoc's 'e.g. leadership moved' names a case that cannot reach this no-op branch

Small, but it misdescribes the fallback contract my ServerCnx comment above is about. The javadoc says the no-op branch covers the case where "the controller is not held locally (e.g. leadership moved)" — and ServerCnx reads the resulting completedFuture(null) as success and removes the per-connection registration ref.

I first read this as a fallback-loss bug, and that reading is wrong; the counter-argument is worth recording: leadership movement does not remove the controllers map entry. Registration installs the future via getOrCreateController (:100-112), a deposed controller stays in the map, and ScalableTopicController.unregisterConsumer's checkLeader() then makes the future complete exceptionally — so ServerCnx takes the error branch and keeps the ref, which is the correct behaviour. Entries are removed only on failed initialization, an explicit releaseController (whose sole production caller is topic deletion), or service shutdown.

So the branch is fine; the example in the javadoc names a case that cannot reach it, which is misleading in a comment whose whole job is to explain when the grace-period fallback still applies. Suggest replacing "e.g. leadership moved" with the cases that actually reach it (topic deleted, service shutting down, or a consumer that never registered here).

* period covers that case.
*/
public CompletableFuture<Void> unregisterConsumer(TopicName topic, String subscription,
String consumerName) {
CompletableFuture<ScalableTopicController> future = controllers.get(topic.toString());
if (future == null) {
return CompletableFuture.completedFuture(null);
}
return future.thenCompose(c -> c.unregisterConsumer(subscription, consumerName));
}

// --- Internal helpers ---

private CompletableFuture<Void> createUnderlyingSegmentTopic(TopicName parentTopic, SegmentInfo segment) {
Expand Down
33 changes: 9 additions & 24 deletions pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -444,21 +444,6 @@ public void testConsumerRejoiningAfterLeaveDoesNotWedgeRelease() throws Exceptio
// watermark) must not leak into the last flip's drain: with everything acked and nothing
// outstanding, a stale delivered-vs-acked pair would make the drain wait forever and the
// rejoining consumer could never attach.
// A consumer's clean close is a disconnect to the controller, which holds its session for
// the grace period before rebalancing. Shrink it (read at coordinator creation, i.e. on
// this subscription's first register) so B1's departure hands the segment back promptly.
int defaultGrace = getPulsar().getConfiguration()
.getScalableTopicConsumerSessionGracePeriodSeconds();
getPulsar().getConfiguration().setScalableTopicConsumerSessionGracePeriodSeconds(1);
try {
runRejoinAfterLeaveScenario();
} finally {
getPulsar().getConfiguration()
.setScalableTopicConsumerSessionGracePeriodSeconds(defaultGrace);
}
}

private void runRejoinAfterLeaveScenario() throws Exception {
String topic = newScalableTopic(1);
admin.scalableTopics().setAutoScalePolicy(topic,
AutoScalePolicyOverride.builder().enabled(false).build());
Expand Down Expand Up @@ -497,12 +482,12 @@ private void runRejoinAfterLeaveScenario() throws Exception {
}
a.acknowledgeCumulative(last.id());

// Phase 2 — B1 joins (shared, individual acks), both drain, then B1 leaves. B1 gets its
// own client: a departure is only visible to the controller as a connection drop, so
// leaving means closing the whole client (the shared client's pooled connection would
// keep B1's registration alive indefinitely).
PulsarClient b1Client = newV5Client();
StreamConsumer<String> b1 = b1Client.newStreamConsumer(Schema.string())
// Phase 2 — B1 joins (shared, individual acks), both drain, then B1 leaves. B1 uses the
// shared client on purpose: its close must reach the controller through the explicit
// unsubscribe (the pooled controller connection stays open, so without it the
// registration would linger for the full disconnect grace period and the segment
// would never be handed back within this test's window).
StreamConsumer<String> b1 = v5Client.newStreamConsumer(Schema.string())
.topic(topic)
.subscriptionName(subscription)
.subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[QUALITY] b1 leaks onto the shared client if an assertion fails before its close

To be clear about what that test does earn: it is a real pin, not a weakened one. The default is one pooled connection per broker and consumer A stays attached, so B1's close cannot make the controller connection inactive and no grace timer can start — the 15-second Awaitility assertion on Exclusive + one consumer is reachable only through the explicit unsubscribe. Good change.

What is left uncovered, all of it behaviour this PR's description advertises:

  • the idempotency contract — an unknown consumer_id must still return CommandSuccess;
  • the ordering guarantee — close() before the subscribe completes, and close-during-reconnect;
  • the broker error path — a failed unregister must still leave close() successful and the connection ref retained;
  • the wire round-trip itself: there is no Commands.newScalableTopicUnsubscribe / PulsarDecoder test at all, so nothing pins the encoding of the new command.

Separately, a small hygiene point on these lines: b1 is neither @Cleanup nor otherwise tracked, and its only close() is after the drain joins and the assertFalse(b1Got.isEmpty(), ...) assertion. Because the shared v5Client is closed only at @AfterClass, an earlier failure leaves B1 registered on the shared client and can contaminate the remaining methods in the class.

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[QUALITY] nothing tests the new command directly, and b1 leaks onto the shared client if an assertion fails before its close

rg 'ScalableTopicUnsubscribe|SCALABLE_TOPIC_UNSUBSCRIBE' over the PR head matches only main sources — ServerCnx, PulsarApi.proto, Commands, PulsarDecoder, ScalableConsumerClient. Zero test references. The only coverage is indirect, through testConsumerRejoiningAfterLeaveDoesNotWedgeRelease, whose actual subject is the stale-watermark wedge.

To be clear about what that test does earn: it is a real pin, not a weakened one. The default is one pooled connection per broker and consumer A stays attached, so B1's close cannot make the controller connection inactive and no grace timer can start — the 15-second Awaitility assertion on Exclusive + one consumer is reachable only through the explicit unsubscribe. Good change.

What is left uncovered, all of it behaviour this PR's description advertises:

  • the idempotency contract — an unknown consumer_id must still return CommandSuccess;
  • the ordering guarantee — close() before the subscribe completes, and close-during-reconnect;
  • the broker error path — a failed unregister must still leave close() successful and the connection ref retained;
  • the wire round-trip itself: there is no Commands.newScalableTopicUnsubscribe / PulsarDecoder test at all, so nothing pins the encoding of the new command.

Separately, a small hygiene point on these lines: b1 is neither @Cleanup nor otherwise tracked, and its only close() is after the drain joins and the assertFalse(b1Got.isEmpty(), ...) assertion. Because the shared v5Client is closed only at @AfterClass, an earlier failure leaves B1 registered on the shared client and can contaminate the remaining methods in the class.

Expand All @@ -515,9 +500,9 @@ private void runRejoinAfterLeaveScenario() throws Exception {
tb1.join();
assertFalse(b1Got.isEmpty(), "consumer B1 received nothing — the segment did not fan out");
b1.close();
b1Client.close();
// Wait until the controller has handed the whole segment back to A and A completed the
// flip back to Exclusive — the stale-watermark state only matters once that is done.
// The clean leave unregisters immediately (no grace wait): the controller hands the
// whole segment back to A, which flips back to Exclusive. Only once that is done does
// the stale-watermark state matter for B2's rejoin.
String segmentTopic = admin.scalableTopics().getStats(topic)
.getSegments().values().iterator().next().name();
Awaitility.await().atMost(Duration.ofSeconds(15)).untilAsserted(() -> {
Expand Down
51 changes: 47 additions & 4 deletions pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,8 @@ final class ScalableConsumerClient implements ScalableConsumerSession, AutoClose
private volatile AssignmentChangeListener listener;
private volatile ClientCnx cnx;
private volatile boolean closed = false;
/** The in-flight (or last) subscribe attempt; close() awaits it before unsubscribing. */
private volatile CompletableFuture<ScalableConsumerAssignment> lastSubscribeResult;

ScalableConsumerClient(PulsarClientImpl v4Client,
TopicName topicName,
Expand Down Expand Up @@ -136,6 +138,11 @@ CompletableFuture<List<ActiveSegment>> start() {
*/
private CompletableFuture<ScalableConsumerAssignment> connectAndSubscribe() {
CompletableFuture<ScalableConsumerAssignment> result = new CompletableFuture<>();
// Published before the closed-flag check below: close() chains its clean unsubscribe on
// this attempt, so either the attempt observes closed and never subscribes, or close()
// observes the attempt and waits for its outcome — never an unsubscribe that overtakes
// the in-flight registration (which would no-op and leave a ghost registration behind).
lastSubscribeResult = result;

DagWatchClient watch = new DagWatchClient(v4Client, topicName);
watch.start()
Expand Down Expand Up @@ -374,11 +381,47 @@ public void close() {
return;
}
closed = true;
CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[BUG] two concurrent reconnect attempts can overwrite lastSubscribeResult, so close() waits on one attempt while another registers afterwards

close() reads a single slot and chains the unsubscribe on it, and connectAndSubscribe() overwrites that slot on every attempt (:145). The ordering argument in the comment above — "either the attempt observes closed and never subscribes, or close() observes the attempt and waits for its outcome" — is a correct Dekker pairing for one in-flight attempt. The existing reconnect paths can produce two.

How two attempts arise. ClientCnx.channelInactive fails all pending requests (ClientCnx.java:535-539) before it invokes session.connectionClosed() (:547). So for an established consumer whose reconnect-subscribe is in flight:

  1. the pending subscribe future is completed exceptionally;
  2. that reaches reconnect()'s .exceptionally handler, which calls scheduleReconnect() (ScalableConsumerClient.java:313-316);
  3. the still-registered session then gets connectionClosed(), which — because initialAssignmentFuture is already done — calls scheduleReconnect() again (:269-283).

Neither the backoff nor initialAssignmentFuture serializes the two, and each connectAndSubscribe() publishes its own result into lastSubscribeResult before its asynchronous lookup/connect.

Failure scenario. Attempt A has already passed its if (closed) check inside thenAccept and is mid-flight when attempt B overwrites the slot. close() waits for B, sends the unsubscribe, and returns; A then completes its registration on the broker afterwards. The result is a registration with no unsubscribe behind it — the zombie group member this PR exists to eliminate, on a pooled connection that never goes inactive. The mutable cnx field compounds it: the unsubscribe is not bound to the connection that performed a particular registration.

The double-scheduleReconnect is pre-existing, but this PR is what makes correctness depend on it not happening. Either fix the double-schedule (have reconnect()'s failure handler not schedule when the failure came from a connection close), or track attempts so close() can chain on all outstanding ones rather than the last.

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[BUG] two concurrent reconnect attempts can overwrite lastSubscribeResult, so close() waits on one attempt while another registers afterwards

close() reads a single slot and chains the unsubscribe on it, and connectAndSubscribe() overwrites that slot on every attempt (:145). The ordering argument in the comment above — "either the attempt observes closed and never subscribes, or close() observes the attempt and waits for its outcome" — is a correct Dekker pairing for one in-flight attempt. The existing reconnect paths can produce two.

How two attempts arise. ClientCnx.channelInactive fails all pending requests (ClientCnx.java:535-539) before it invokes session.connectionClosed() (:547). So for an established consumer whose reconnect-subscribe is in flight:

  1. the pending subscribe future is completed exceptionally;
  2. that reaches reconnect()'s .exceptionally handler, which calls scheduleReconnect() (ScalableConsumerClient.java:313-316);
  3. the still-registered session then gets connectionClosed(), which — because initialAssignmentFuture is already done — calls scheduleReconnect() again (:269-283).

Neither the backoff nor initialAssignmentFuture serializes the two, and each connectAndSubscribe() publishes its own result into lastSubscribeResult before its asynchronous lookup/connect.

Failure scenario. Attempt A has already passed its if (closed) check inside thenAccept and is mid-flight when attempt B overwrites the slot. close() waits for B, sends the unsubscribe, and returns; A then completes its registration on the broker afterwards. The result is a registration with no unsubscribe behind it — the zombie group member this PR exists to eliminate, on a pooled connection that never goes inactive. The mutable cnx field compounds it: the unsubscribe is not bound to the connection that performed a particular registration.

The double-scheduleReconnect is pre-existing, but this PR is what makes correctness depend on it not happening. Either fix the double-schedule (have reconnect()'s failure handler not schedule when the failure came from a connection close), or track attempts so close() can chain on all outstanding ones rather than the last.

if (pending == null) {
// Never attempted to subscribe: nothing registered anywhere.
return;
}
// Clean leave: tell the controller to unregister this consumer and rebalance the group
// immediately, instead of holding the registration for the disconnect grace period (the
// controller connection is pooled, so closing this consumer does not close the channel
// and the broker would otherwise never notice the departure while the client lives).
// Chained on the in-flight subscribe outcome so the unsubscribe can never overtake the
// registration (the broker records it before sending the subscribe response). Sent even
// when the subscribe failed — the command is idempotent, and skipping on a razor-edge
// failure would risk leaving a registration behind. Best-effort throughout: on any
// failure the grace period remains the fallback for the eventual real disconnect.
pending.whenComplete((__, subscribeEx) -> sendUnsubscribe());

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[BUG] close() now defers removeScalableConsumerSession behind a subscribe future that has no timeout, so an unanswered subscribe retains the closed session for the life of the pooled connection

Before this PR, close() called c.removeScalableConsumerSession(consumerId) synchronously. It now happens only inside sendUnsubscribe(), which is reached from pending.whenComplete(...) — so both the wire command and the local unregistration wait on the subscribe future.

That future has no timeout. The subscribe response future is put straight into the map at ScalableConsumerClient.java:195-197 — cnx.getPendingRequests().put(requestId, responseFuture) — with no RequestTime added to ClientCnx.requestTimeoutQueue, and ClientCnx.checkRequestTimeout() walks only that queue, never pendingRequests. TimedCompletableFuture carries no timer of its own. So the only thing that can ever complete it is the connection-close sweep at ClientCnx.java:535-539.

Failure scenario. A live pooled connection where the broker accepts the subscribe and never answers it (for example registerConsumer's future stalls on the metadata store). The application calls close(); pending never completes; removeScalableConsumerSession is never called; the closed consumer stays in ClientCnx.scalableConsumerSessions for the life of the pooled connection, keeps a strong reference to the ScalableConsumerClient, and keeps ClientCnx.idleCheck() (:2152-2175) reporting the connection as non-idle so the pool cannot reap it. No unsubscribe is ever sent either.

The local unregistration does not need to wait on anything — the subscribe response is routed through pendingRequests, not through scalableConsumerSessions — so close() can do removeScalableConsumerSession unconditionally (as it did before) and defer only the wire command.

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[BUG] close() now defers removeScalableConsumerSession behind a subscribe future that has no timeout, so an unanswered subscribe retains the closed session for the life of the pooled connection

Before this PR, close() called c.removeScalableConsumerSession(consumerId) synchronously. It now happens only inside sendUnsubscribe(), which is reached from pending.whenComplete(...) — so both the wire command and the local unregistration wait on the subscribe future.

That future has no timeout. The subscribe response future is put straight into the map at ScalableConsumerClient.java:195-197 — cnx.getPendingRequests().put(requestId, responseFuture) — with no RequestTime added to ClientCnx.requestTimeoutQueue, and ClientCnx.checkRequestTimeout() walks only that queue, never pendingRequests. TimedCompletableFuture carries no timer of its own. So the only thing that can ever complete it is the connection-close sweep at ClientCnx.java:535-539.

Failure scenario. A live pooled connection where the broker accepts the subscribe and never answers it (for example registerConsumer's future stalls on the metadata store). The application calls close(); pending never completes; removeScalableConsumerSession is never called; the closed consumer stays in ClientCnx.scalableConsumerSessions for the life of the pooled connection, keeps a strong reference to the ScalableConsumerClient, and keeps ClientCnx.idleCheck() (:2152-2175) reporting the connection as non-idle so the pool cannot reap it. No unsubscribe is ever sent either.

The local unregistration does not need to wait on anything — the subscribe response is routed through pendingRequests, not through scalableConsumerSessions — so close() can do removeScalableConsumerSession unconditionally (as it did before) and defer only the wire command.

}

private void sendUnsubscribe() {
ClientCnx c = cnx;
if (c != null) {
c.removeScalableConsumerSession(consumerId);
// No close command for now — broker reaps registrations via grace timer on
// disconnect. A future refactor can add an explicit unsubscribe.
if (c == null) {
return;
}
c.removeScalableConsumerSession(consumerId);
try {
long requestId = v4Client.newRequestId();
var responseFuture = new TimedCompletableFuture<Void>();
c.getPendingRequests().put(requestId, responseFuture);
c.ctx().writeAndFlush(Commands.newScalableTopicUnsubscribe(requestId, consumerId))
.addListener(writeFuture -> {
if (!writeFuture.isSuccess()) {
c.getPendingRequests().remove(requestId);
}
});

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[QUALITY] the unsubscribe request bypasses ClientCnx's timeout bookkeeping and pins the pooled connection as non-idle; its write-failure path never completes the future it logs from

The unsubscribe request duplicates ClientCnx's request bookkeeping instead of using it, in three small ways:

  1. No timeout entry. getPendingRequests().put(requestId, responseFuture) with no RequestTime added to requestTimeoutQueue — compare ClientCnx.sendRequestAndHandleTimeout (ClientCnx.java:1471-1487), which always adds one. An unanswered unsubscribe therefore sits in pendingRequests for the life of the connection, and ClientCnx.idleCheck() (:2152-2155) returns false while that map is non-empty — so a single unanswered unsubscribe keeps the pooled connection from ever being reaped as idle.
  2. The write-failure listener never completes the future. On !writeFuture.isSuccess() it removes the map entry but leaves responseFuture pending forever, so the .exceptionally(...) logger installed three lines below cannot fire on the one failure it exists to report. The subscribe path in the same class does complete its result on write failure.
  3. A third point I had drafted here does not hold, for the record — using getPendingRequests().remove(requestId) rather than the removePendingRequest(requestId, expectedFuture) CAS primitive is not independently unsafe here (request ids are globally unique and this is not a lookup request holding a permit), so treat that as style, not a defect.

The pre-existing subscribe path in this class has the same shape, so a small ClientCnx helper for "send a scalable-session request with the standard timeout bookkeeping" would fix both at once.

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[QUALITY] the unsubscribe request bypasses ClientCnx's timeout bookkeeping and pins the pooled connection as non-idle; its write-failure path never completes the future it logs from

The unsubscribe request duplicates ClientCnx's request bookkeeping instead of using it, in three small ways:

  1. No timeout entry. getPendingRequests().put(requestId, responseFuture) with no RequestTime added to requestTimeoutQueue — compare ClientCnx.sendRequestAndHandleTimeout (ClientCnx.java:1471-1487), which always adds one. An unanswered unsubscribe therefore sits in pendingRequests for the life of the connection, and ClientCnx.idleCheck() (:2152-2155) returns false while that map is non-empty — so a single unanswered unsubscribe keeps the pooled connection from ever being reaped as idle.
  2. The write-failure listener never completes the future. On !writeFuture.isSuccess() it removes the map entry but leaves responseFuture pending forever, so the .exceptionally(...) logger installed three lines below cannot fire on the one failure it exists to report. The subscribe path in the same class does complete its result on write failure.
  3. A third point I had drafted here does not hold, for the record — using getPendingRequests().remove(requestId) rather than the removePendingRequest(requestId, expectedFuture) CAS primitive is not independently unsafe here (request ids are globally unique and this is not a lookup request holding a permit), so treat that as style, not a defect.

The pre-existing subscribe path in this class has the same shape, so a small ClientCnx helper for "send a scalable-session request with the standard timeout bookkeeping" would fix both at once.

responseFuture.exceptionally(ex -> {
log.debug().exceptionMessage(ex)
.log("Clean unsubscribe failed; relying on the grace period");
return null;
});
} catch (Exception e) {
log.debug().exceptionMessage(e)
.log("Clean unsubscribe failed; relying on the grace period");
}
}

Expand Down
13 changes: 13 additions & 0 deletions pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1804,6 +1804,19 @@ public static ByteBuf newScalableTopicSubscribe(long requestId, String topic,
return serializeWithSize(cmd);
}

/**
* Client -> Broker: a scalable consumer is cleanly leaving its subscription; the
* controller unregisters it and rebalances immediately instead of waiting out the
* disconnect grace period. Acknowledged with {@code CommandSuccess}.
*/
public static ByteBuf newScalableTopicUnsubscribe(long requestId, long consumerId) {
BaseCommand cmd = localCmd(Type.SCALABLE_TOPIC_UNSUBSCRIBE);
cmd.setScalableTopicUnsubscribe()
.setRequestId(requestId)
.setConsumerId(consumerId);
return serializeWithSize(cmd);
}

/**
* Broker -> Client: response to a scalable-topic subscribe request. On success the
* caller must populate the nested {@link ScalableConsumerAssignment} via
Expand Down
11 changes: 11 additions & 0 deletions pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,7 @@
import org.apache.pulsar.common.api.proto.CommandScalableTopicLookup;
import org.apache.pulsar.common.api.proto.CommandScalableTopicSubscribe;
import org.apache.pulsar.common.api.proto.CommandScalableTopicSubscribeResponse;
import org.apache.pulsar.common.api.proto.CommandScalableTopicUnsubscribe;
import org.apache.pulsar.common.api.proto.CommandScalableTopicUpdate;
import org.apache.pulsar.common.api.proto.CommandSeek;
import org.apache.pulsar.common.api.proto.CommandSend;
Expand Down Expand Up @@ -514,6 +515,11 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
handleCommandScalableTopicAssignmentUpdate(cmd.getScalableTopicAssignmentUpdate());
break;

case SCALABLE_TOPIC_UNSUBSCRIBE:
checkArgument(cmd.hasScalableTopicUnsubscribe());
handleCommandScalableTopicUnsubscribe(cmd.getScalableTopicUnsubscribe());
break;

case WATCH_SCALABLE_TOPICS:
checkArgument(cmd.hasWatchScalableTopics());
handleCommandWatchScalableTopics(cmd.getWatchScalableTopics());
Expand Down Expand Up @@ -837,6 +843,11 @@ protected void handleCommandScalableTopicAssignmentUpdate(
throw new UnsupportedOperationException();
}

protected void handleCommandScalableTopicUnsubscribe(
CommandScalableTopicUnsubscribe commandScalableTopicUnsubscribe) {
throw new UnsupportedOperationException();
}

protected void handleCommandWatchScalableTopics(
org.apache.pulsar.common.api.proto.CommandWatchScalableTopics commandWatchScalableTopics) {
throw new UnsupportedOperationException();
Expand Down
14 changes: 14 additions & 0 deletions pulsar-common/src/main/proto/PulsarApi.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -995,6 +995,16 @@ message CommandScalableTopicAssignmentUpdate {
required ScalableConsumerAssignment assignment = 2;
}

// Client -> Broker: a scalable consumer is cleanly leaving its subscription. The controller
// deletes the registration and rebalances the group immediately, instead of holding the
// session for the disconnect grace period (which remains the fallback for unclean
// departures). Acknowledged with CommandSuccess; idempotent — an unknown consumer_id (e.g.
// already swept by a disconnect) still succeeds.
message CommandScalableTopicUnsubscribe {
required uint64 request_id = 1;
required uint64 consumer_id = 2;
}

// Multi-topic consumer watcher: subscribes to the union of scalable topics in a
// namespace that match a (possibly empty) set of property filters. The broker keeps
// pushing updates as topics enter or leave the matching set. See
Expand Down Expand Up @@ -1356,6 +1366,8 @@ message BaseCommand {
WATCH_TC_ASSIGNMENTS = 79;
WATCH_TC_ASSIGNMENTS_UPDATE = 80;
WATCH_TC_ASSIGNMENTS_CLOSE = 81;

SCALABLE_TOPIC_UNSUBSCRIBE = 82;
}


Expand Down Expand Up @@ -1455,4 +1467,6 @@ message BaseCommand {
optional CommandWatchTcAssignments watchTcAssignments = 79;
optional CommandWatchTcAssignmentsUpdate watchTcAssignmentsUpdate = 80;
optional CommandWatchTcAssignmentsClose watchTcAssignmentsClose = 81;

optional CommandScalableTopicUnsubscribe scalableTopicUnsubscribe = 82;
}
Loading