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

["merlimat (via GitHub)" <gi...@apache.org>]

merlimat opened a new pull request, #26433:
URL: https://github.com/apache/pulsar/pull/26433

   ### Motivation
   
   Follow-up promised in [#26208](https://github.com/apache/pulsar/pull/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.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on code in PR #26433:
URL: https://github.com/apache/pulsar/pull/26433#discussion_r3894618269


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -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.

Review Comment:
   **[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:
   
   ```java
   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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());

Review Comment:
   **[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.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java:
##########
@@ -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

Review Comment:
   **[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).



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());
+    }
+
+    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);
+                        }
+                    });

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;

Review Comment:
   **[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.



##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java:
##########
@@ -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)

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());
+    }
+
+    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);
+                        }
+                    });

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;

Review Comment:
   **[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.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -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.

Review Comment:
   **[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:
   
   ```java
   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.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java:
##########
@@ -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

Review Comment:
   **[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).



##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java:
##########
@@ -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)

Review Comment:
   **[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.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on code in PR #26433:
URL: https://github.com/apache/pulsar/pull/26433#discussion_r3887548750


##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());
+    }
+
+    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);
+                        }
+                    });

Review Comment:
   **[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.



##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java:
##########
@@ -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)

Review Comment:
   **[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.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java:
##########
@@ -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

Review Comment:
   **[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).



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -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.

Review Comment:
   **[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:
   
   ```java
   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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());
+    }
+
+    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);
+                        }
+                    });

Review Comment:
   **[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.



##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java:
##########
@@ -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)

Review Comment:
   **[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.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java:
##########
@@ -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

Review Comment:
   **[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).



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on code in PR #26433:
URL: https://github.com/apache/pulsar/pull/26433#discussion_r3887550433


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -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.

Review Comment:
   **[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:
   
   ```java
   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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());

Review Comment:
   **[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.



##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java:
##########
@@ -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)

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());
+    }
+
+    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);
+                        }
+                    });

Review Comment:
   **[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.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java:
##########
@@ -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

Review Comment:
   **[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).



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["merlimat (via GitHub)" <gi...@apache.org>]

merlimat commented on code in PR #26433:
URL: https://github.com/apache/pulsar/pull/26433#discussion_r3919795679


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -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) {

Review Comment:
   Confirmed — and your framing was exactly right: the round-2 timeout is what opened the window. Fixed in f75a7d0cfaa: the registration ref is recorded **before** `registerConsumer` resolves and carries the registration future; both the unsubscribe handler and the `channelInactive` sweep chain behind it, so the ordering no longer depends on how long the client could wait (a failed registration drops the ref, and the chained unregister is idempotent over it). Chaining the sweep also closes the adjacent pre-existing corner where a registration completing after its connection died never got a disconnect report. Both builders now close the session on their failure paths, so an abandoned subscribe still sends its unsubscribe once the in-flight attempt settles.
   
   New ServerCnx test pins the ordering: an unsubscribe sent mid-registration answers nothing until the registration completes, then runs the unregister and answers success.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinator.java:
##########
@@ -214,19 +214,25 @@ public synchronized CompletableFuture<Map<ConsumerSession, ConsumerAssignment>>
     }
 
     /**
-     * Explicit unregister (consumer asked to leave the subscription). Cancels any pending
-     * grace timer, deletes the persisted registration, and rebalances.
+     * Explicit unregister (consumer asked to leave the subscription). Deletes the persisted
+     * registration first, and only on success removes the in-memory session, cancels its
+     * grace timer, and rebalances. A failed delete therefore changes nothing: the session
+     * stays registered and connected, so the channelInactive → grace-period fallback still
+     * works, and a retried unregister actually retries the deletion instead of short-
+     * circuiting on an already-removed session.
      */
     public synchronized CompletableFuture<Map<ConsumerSession, ConsumerAssignment>> unregisterConsumer(
             String consumerName) {
-        ConsumerSession removed = sessions.remove(consumerName);
-        if (removed == null) {
+        if (!sessions.containsKey(consumerName)) {
             return CompletableFuture.completedFuture(snapshotAssignments());
         }
-        removed.cancelGraceTimer();
         return resources.unregisterConsumerAsync(topicName, subscriptionName, consumerName)
                 .thenApply(__ -> {
                     synchronized (this) {
+                        ConsumerSession removed = sessions.remove(consumerName);

Review Comment:
   Confirmed, implemented as you suggested in f75a7d0cfaa: the leaving consumer's id (already the key of the connection's registration map) is plumbed down through `ScalableTopicService`/`ScalableTopicController`, and the delete's continuation removes the session only if its current `getConsumerId()` still matches — a rejoin that won the race keeps its session, and the persisted registration the delete just erased is restored so a controller failover still knows the member.
   
   Covered by a coordinator identity test (stale-id leave keeps the rejoined session; live-id leave removes it) and a same-name close/resubscribe churn leg added to the e2e — thanks for pointing out the generated-name blind spot there.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on code in PR #26433:
URL: https://github.com/apache/pulsar/pull/26433#discussion_r3887551142


##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;

Review Comment:
   **[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.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -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.

Review Comment:
   **[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:
   
   ```java
   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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());
+    }
+
+    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);
+                        }
+                    });

Review Comment:
   **[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.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java:
##########
@@ -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

Review Comment:
   **[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).



##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java:
##########
@@ -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)

Review Comment:
   **[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.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["merlimat (via GitHub)" <gi...@apache.org>]

merlimat merged PR #26433:
URL: https://github.com/apache/pulsar/pull/26433


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on code in PR #26433:
URL: https://github.com/apache/pulsar/pull/26433#discussion_r3887550433


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -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.

Review Comment:
   **[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:
   
   ```java
   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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());
+    }
+
+    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);
+                        }
+                    });

Review Comment:
   **[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.



##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java:
##########
@@ -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)

Review Comment:
   **[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.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java:
##########
@@ -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

Review Comment:
   **[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).



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -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.

Review Comment:
   **[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:
   
   ```java
   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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());
+    }
+
+    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);
+                        }
+                    });

Review Comment:
   **[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.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on code in PR #26433:
URL: https://github.com/apache/pulsar/pull/26433#discussion_r3887548745


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -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.

Review Comment:
   **[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:
   
   ```java
   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.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on code in PR #26433:
URL: https://github.com/apache/pulsar/pull/26433#discussion_r3887550829


##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());

Review Comment:
   **[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.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -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.

Review Comment:
   **[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:
   
   ```java
   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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());
+    }
+
+    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);
+                        }
+                    });

Review Comment:
   **[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.



##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java:
##########
@@ -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)

Review Comment:
   **[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.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java:
##########
@@ -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

Review Comment:
   **[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).



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on code in PR #26433:
URL: https://github.com/apache/pulsar/pull/26433#discussion_r3887550438


##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;

Review Comment:
   **[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.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on code in PR #26433:
URL: https://github.com/apache/pulsar/pull/26433#discussion_r3916837738


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinator.java:
##########
@@ -214,19 +214,25 @@ public synchronized CompletableFuture<Map<ConsumerSession, ConsumerAssignment>>
     }
 
     /**
-     * Explicit unregister (consumer asked to leave the subscription). Cancels any pending
-     * grace timer, deletes the persisted registration, and rebalances.
+     * Explicit unregister (consumer asked to leave the subscription). Deletes the persisted
+     * registration first, and only on success removes the in-memory session, cancels its
+     * grace timer, and rebalances. A failed delete therefore changes nothing: the session
+     * stays registered and connected, so the channelInactive → grace-period fallback still
+     * works, and a retried unregister actually retries the deletion instead of short-
+     * circuiting on an already-removed session.
      */
     public synchronized CompletableFuture<Map<ConsumerSession, ConsumerAssignment>> unregisterConsumer(
             String consumerName) {
-        ConsumerSession removed = sessions.remove(consumerName);
-        if (removed == null) {
+        if (!sessions.containsKey(consumerName)) {
             return CompletableFuture.completedFuture(snapshotAssignments());
         }
-        removed.cancelGraceTimer();
         return resources.unregisterConsumerAsync(topicName, subscriptionName, consumerName)
                 .thenApply(__ -> {
                     synchronized (this) {
+                        ConsumerSession removed = sessions.remove(consumerName);

Review Comment:
   **[QUALITY] a same-name rejoin during the in-flight delete is removed again by the delete's own continuation**
   
   The reorder is the right fix for the failed-delete case. It leaves one window on the success path: while the delete is in flight the session is still in `sessions`, so a `registerConsumer` for the same `consumerName` takes the reconnect branch — it attaches the new connection and consumer id, cancels the grace timer, and answers with an assignment:
   
   https://github.com/apache/pulsar/blob/7ac6b40b153d6cf8844a19c2c7b1d11e59d1c7da/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinator.java#L186-L198
   
   When the delete then completes, this line removes that same session object, cancels its (now freshly attached) timer and rebalances its segments to the others. The rejoined consumer holds an assignment the coordinator has forgotten, and the persisted registration is gone as well — the reconnect branch does not re-create it — so a controller failover will not restore it either.
   
   Reachable whenever an application closes a consumer and immediately re-subscribes under the same `consumerName`, which the builder lets you set explicitly (`StreamConsumerBuilderV5.consumerName`). `testRapidSubscribeCloseChurnLeavesGroupClean` does not cover it: each iteration builds a fresh builder, so every churned consumer gets its own generated `v5-stream-<random>` name.
   
   The previous ordering had its own version of this race — the rejoin took the fresh-registration path and the in-flight delete could remove the znode it had just written — so this is not a regression, just the last corner of the same problem, and the new ordering makes it visible immediately rather than at the next failover. Guarding the removal on identity closes it: only remove when the session is still the one the unregister observed, e.g. by carrying the unsubscribe's `consumerId` down from `ServerCnx` — it is already the key of `scalableConsumerRegistrations` — and comparing it against the session's current `getConsumerId()`, which `attach` overwrites.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -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) {

Review Comment:
   **[BUG] the new subscribe timeout lets the unsubscribe overtake a registration that is still in flight, and this branch then answers success without unregistering it**
   
   The registration ref is recorded only *after* `registerConsumer` resolves ([`ServerCnx.java:1152-1166`](https://github.com/apache/pulsar/blob/7ac6b40b153d6cf8844a19c2c7b1d11e59d1c7da/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java#L1152-L1166)), so for the whole duration of the registration this map has no entry for the consumer and an unsubscribe arriving in that window takes the branch above: `CommandSuccess`, nothing unregistered.
   
   Until this round the client could not reach that window. The subscribe future had no timeout, so it completed only on a response or on the connection dying — and a dying connection is swept by `channelInactive`. Now the subscribe goes through `sendScalableSessionRequest`, which registers it in `requestTimeoutQueue`, so it can also complete exceptionally after `operationTimeoutMs` **while the connection stays up** ([`ClientCnx.java:2139-2152`](https://github.com/apache/pulsar/blob/7ac6b40b153d6cf8844a19c2c7b1d11e59d1c7da/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java#L2139-L2152)). The attempt then drops straight out of `inFlightSubscribes`, which is what `close()` waits on:
   
   https://github.com/apache/pulsar/blob/7ac6b40b153d6cf8844a19c2c7b1d11e59d1c7da/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java#L217-L227
   
   **Failure scenario.** A registration whose metadata write outlives `operationTimeoutMs` (30s by default, and often configured lower). The client gives up on the subscribe; the broker completes the registration afterwards. Two ways that ends badly:
   
   - the application calls `close()` — the unsubscribe is sent, finds no ref yet, and is answered as an idempotent success;
   - or nothing is sent at all, because a failed initial subscribe never closes the session ([`ScalableConsumerClient.java:127-141`](https://github.com/apache/pulsar/blob/7ac6b40b153d6cf8844a19c2c7b1d11e59d1c7da/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java#L127-L141)): both consumer types close it when *their own* creation fails, but neither builder closes it when `start()` itself fails ([`StreamConsumerBuilderV5.java:98-100`](https://github.com/apache/pulsar/blob/7ac6b40b153d6cf8844a19c2c7b1d11e59d1c7da/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/StreamConsumerBuilderV5.java#L98-L100), [`CheckpointConsumerBuilderV5.java:81-85`](https://github.com/apache/pulsar/blob/7ac6b40b153d6cf8844a19c2c7b1d11e59d1c7da/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/CheckpointConsumerBuilderV5.java#L81-L85)).
   
   Either way the group keeps a member with no client behind it, on a pooled connection that never goes inactive — the wedge this PR exists to remove.
   
   Two containments, and the first is the one that matters: record the ref (or a placeholder for it) **before** awaiting `registerConsumer` and drop it if the registration fails, so an unsubscribe arriving mid-registration is ordered behind it instead of silently no-oping — that makes the guarantee independent of how long the client is able to wait. Second, have the builder close the session on its failure paths so an abandoned subscribe always sends its unsubscribe.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java:
##########
@@ -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

Review Comment:
   Confirmed at 7ac6b40b153 — the example is replaced by the cases that actually reach the branch, and the deposed-leader path is stated rather than left to be re-derived. Thanks for encoding the trace; that was the part I most wanted written down.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -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.

Review Comment:
   Confirmed at 7ac6b40b153 — the delete is now the thing that gates every mutation:
   
   https://github.com/apache/pulsar/blob/7ac6b40b153d6cf8844a19c2c7b1d11e59d1c7da/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinator.java#L226-L242
   
   So a failed delete leaves the session registered *and* connected, which is exactly what the comment on this line promises: `ServerCnx` keeps the ref, `channelInactive` still reaches `onConsumerDisconnect`, and the grace timer is armed there rather than having been cancelled already. The `containsKey` guard also fixes the second half of the point — a retry after a failed delete now re-runs the deletion instead of short-circuiting on an absent session — and `testScalableTopicUnsubscribeIdempotencyAndErrorPath` pins that sequence end to end.
   
   One narrow corner of the new ordering is worth a look; I have raised it separately on [`SubscriptionCoordinator.java:232`](https://github.com/apache/pulsar/blob/7ac6b40b153d6cf8844a19c2c7b1d11e59d1c7da/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinator.java#L232) rather than reopening this thread.



##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java:
##########
@@ -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)

Review Comment:
   Confirmed at 7ac6b40b153: the wire round-trip, the idempotency contract and the error/retry path all have direct tests now, and `b1` is tracked. The churn e2e is a good choice of shape — with the default 60s grace, a missed unregistration cannot be hidden inside the 20s assertion window.
   
   Your note about the deterministic close-mid-subscribe case is fair, and the attempt-set invariant is the right thing to lean on there. One gap to be aware of rather than to fix in this test: every consumer in the churn loop gets its own generated `v5-stream-<random>` name, because each iteration builds a fresh builder — so the rejoin-under-the-same-name sequence is not exercised. That is the case behind the comment on [`SubscriptionCoordinator.java:232`](https://github.com/apache/pulsar/blob/7ac6b40b153d6cf8844a19c2c7b1d11e59d1c7da/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinator.java#L232).



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on PR #26433:
URL: https://github.com/apache/pulsar/pull/26433#issuecomment-5516207426

   Feel free to mark the unresolved threads as resolved so that this PR can be merged. No blockers.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on code in PR #26433:
URL: https://github.com/apache/pulsar/pull/26433#discussion_r3887550835


##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java:
##########
@@ -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)

Review Comment:
   **[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.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java:
##########
@@ -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

Review Comment:
   **[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).



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -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.

Review Comment:
   **[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:
   
   ```java
   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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());
+    }
+
+    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);
+                        }
+                    });

Review Comment:
   **[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.



##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java:
##########
@@ -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)

Review Comment:
   **[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.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java:
##########
@@ -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

Review Comment:
   **[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).



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on code in PR #26433:
URL: https://github.com/apache/pulsar/pull/26433#discussion_r3887550006


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -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.

Review Comment:
   **[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:
   
   ```java
   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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());

Review Comment:
   **[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.



##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java:
##########
@@ -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)

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());
+    }
+
+    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);
+                        }
+                    });

Review Comment:
   **[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.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java:
##########
@@ -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

Review Comment:
   **[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).



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on code in PR #26433:
URL: https://github.com/apache/pulsar/pull/26433#discussion_r3916840549


##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;

Review Comment:
   Confirmed at 7ac6b40b153. Both halves are there: `inFlightSubscribes` replaces the overwritable slot, with each attempt added before its closed-check so the pairing you describe holds per attempt, and `close()` waits on `allOf` over the set.
   
   I checked the CAS against the ordering that produced the double schedule in the first place, since the flag is cleared at the *start* of `reconnect()` rather than when the attempt settles: `channelInactive` fails the pending requests ([`ClientCnx.java:535-547`](https://github.com/apache/pulsar/blob/7ac6b40b153d6cf8844a19c2c7b1d11e59d1c7da/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java#L535-L547)) before it calls `connectionClosed()`, so the request-failure path reaches `scheduleReconnect` first, wins the CAS and arms a timer at least one backoff away — and the `connectionClosed` call, which is synchronous with it, then finds the flag still set and drops out. The attempt that a session's `connectionClosed` could report on is always the one holding that connection, so I could not construct a second trigger arriving while an attempt is still running.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());

Review Comment:
   Confirmed at 7ac6b40b153 ([`ScalableConsumerClient.java:391-421`](https://github.com/apache/pulsar/blob/7ac6b40b153d6cf8844a19c2c7b1d11e59d1c7da/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java#L391-L421)): the local deregistration is back to being immediate and unconditional, and only the wire command waits. Worth noting that this also stays correct for the attempt that registers on a *different* connection after `close()` ran, because `sendUnsubscribe()` re-reads `cnx` rather than reusing the one `close()` captured.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());
+    }
+
+    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);
+                        }
+                    });

Review Comment:
   Confirmed at 7ac6b40b153 — one helper, both call sites, and the two concrete defects (no `requestTimeoutQueue` entry, a write-failure path that never completed the future it logged from) are gone from the subscribe as well as the unsubscribe.
   
   The one consequence worth naming is that the subscribe future can now complete exceptionally while the connection is still alive, which the close-ordering argument did not have to consider before. I have raised that on [`ServerCnx.java:1192`](https://github.com/apache/pulsar/blob/7ac6b40b153d6cf8844a19c2c7b1d11e59d1c7da/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java#L1192) rather than here, since the containment belongs on the broker side.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari commented on code in PR #26433:
URL: https://github.com/apache/pulsar/pull/26433#discussion_r3887548745


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -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.

Review Comment:
   **[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:
   
   ```java
   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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;

Review Comment:
   **[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.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java:
##########
@@ -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

Review Comment:
   **[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).



##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java:
##########
@@ -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)

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());
+    }
+
+    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);
+                        }
+                    });

Review Comment:
   **[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.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());

Review Comment:
   **[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.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][client] PIP-486: explicit unsubscribe for scalable consumers' clean close [pulsar]

["merlimat (via GitHub)" <gi...@apache.org>]

merlimat commented on code in PR #26433:
URL: https://github.com/apache/pulsar/pull/26433#discussion_r3899450527


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -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.

Review Comment:
   Confirmed. Fixed in 7ac6b40b153: the coordinator now deletes the persisted registration **first** and mutates nothing until it succeeds — the session removal, grace-timer cancel, and rebalance all live in the success path. A failed delete therefore leaves the session registered and connected (channelInactive → grace fallback works as the ServerCnx comment promises), and a retried unsubscribe genuinely retries the deletion instead of short-circuiting on an already-removed session. Covered by a new ServerCnx test: error response keeps the ref, the retry re-runs the unregister, success removes it.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;

Review Comment:
   Confirmed, including the double-`scheduleReconnect` chain. Two changes in 7ac6b40b153: reconnect scheduling is collapsed to a single pending attempt (`reconnectPending` CAS, cleared when the attempt starts), and `close()` now chains the unsubscribe on **every** in-flight subscribe attempt — a set replaces the single last-attempt slot, with each attempt registered before its closed-check so the per-attempt Dekker pairing holds for all of them. Settled attempts drop out of the set; their registrations, if any, already exist and are covered by the trailing unsubscribe.



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());

Review Comment:
   Confirmed — restored in 7ac6b40b153: `close()` calls `removeScalableConsumerSession` immediately and unconditionally (as before the clean leave existed); only the wire command waits on the in-flight attempts. The never-completing wait is also gone at the source, since the subscribe now carries standard request-timeout tracking (see the bookkeeping thread).



##########
pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java:
##########
@@ -374,11 +381,47 @@ public void close() {
             return;
         }
         closed = true;
+        CompletableFuture<ScalableConsumerAssignment> pending = lastSubscribeResult;
+        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());
+    }
+
+    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);
+                        }
+                    });

Review Comment:
   Done as suggested in 7ac6b40b153: a `ClientCnx.sendScalableSessionRequest` helper delegating to the existing `sendRequestAndHandleTimeout` (pending-request registration, `requestTimeoutQueue` entry, write-failure completion), now used by both the subscribe and the unsubscribe — the hand-rolled entries that never timed out and the write-failure path that left the future forever pending are gone from both. Thanks also for recording the retraction on the CAS-remove point.



##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java:
##########
@@ -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)

Review Comment:
   Added in 7ac6b40b153: a `Commands` round-trip test for the new command; a ServerCnx test pinning the idempotency contract (unknown `consumer_id` → `CommandSuccess`) and the error path (error response → ref retained → a retry re-runs the unregister → success removes it, then idempotent success again); and a subscribe/close churn e2e — five join-and-close cycles against the default 60s grace, after which the group must converge back to one Exclusive owner, which any ghost registration would prevent. `b1` is now tracked for cleanup.
   
   One honest gap: a *deterministic* close-mid-subscribe test isn't reachable through the public API (the consumer object only exists once the subscribe completes). That window is covered by the attempt-set invariant from the other thread — close chains on all in-flight attempts, each registered before its closed-check — and exercised probabilistically by the churn e2e.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java:
##########
@@ -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

Review Comment:
   Fixed in 7ac6b40b153 — the javadoc now names the cases that actually reach the no-op branch (deleted topic, shutdown, never-registered consumer) and records the deposed-leader behaviour you traced: the controller entry survives leadership loss and fails via `checkLeader()` into the ref-preserving error path. Thanks for working through and documenting your own initial misreading — that trace was useful to encode.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org