From dddb4a6f7f2cd9393f63d269ce9597ca8f1241ea Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Fri, 7 Aug 2026 21:43:15 +0200 Subject: [PATCH 1/2] [fix][client] Scalable consumers: explicit unsubscribe on clean close MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A scalable consumer's registration lives on the controller, but its connection comes from the shared connection pool: a consumer's close never makes the channel inactive while the client process lives, so the controller never noticed a clean departure — the registration stayed "connected" indefinitely (a same-process leave was a permanent zombie group member) and even a cross-process clean leave stalled the rebalance for the full disconnect grace period. Adds CommandScalableTopicUnsubscribe (BaseCommand type 82): on close the client tells the controller to delete the registration and rebalance the group immediately; the disconnect grace period remains the fallback for unclean departures. - Client: the unsubscribe is chained on the in-flight subscribe attempt's outcome, so it can never overtake the registration (the broker records the per-connection registration ref before sending the subscribe response); best-effort — a failure never fails the close. - Broker: the handler resolves the consumer through the connection's own registration map (so a client can only unregister sessions it created), forwards to the existing SubscriptionCoordinator.unregisterConsumer (grace-timer cancel + persisted-entry delete + rebalance), and answers CommandSuccess, idempotently for unknown ids. The registration ref is removed only after the unregister succeeds, keeping the channelInactive grace fallback alive if the explicit path fails. The rejoin e2e now runs against the default grace period with all consumers on one shared client: the leave hands the segment back within seconds purely through the unsubscribe path (previously it required a dedicated client per leaver and a shrunken grace period). Assisted-by: Claude Code (Fable 5) --- .../pulsar/broker/service/ServerCnx.java | 40 +++++++++++++++ .../scalable/ScalableTopicService.java | 15 ++++++ .../api/v5/V5EntryBucketDispatchTest.java | 33 ++++-------- .../impl/v5/ScalableConsumerClient.java | 51 +++++++++++++++++-- .../pulsar/common/protocol/Commands.java | 13 +++++ .../pulsar/common/protocol/PulsarDecoder.java | 11 ++++ pulsar-common/src/main/proto/PulsarApi.proto | 14 +++++ 7 files changed, 149 insertions(+), 28 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index 5e8422af870f4..ac600b047876d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java @@ -142,6 +142,7 @@ import org.apache.pulsar.common.api.proto.CommandScalableTopicClose; import org.apache.pulsar.common.api.proto.CommandScalableTopicLookup; import org.apache.pulsar.common.api.proto.CommandScalableTopicSubscribe; +import org.apache.pulsar.common.api.proto.CommandScalableTopicUnsubscribe; import org.apache.pulsar.common.api.proto.CommandSeek; import org.apache.pulsar.common.api.proto.CommandSend; import org.apache.pulsar.common.api.proto.CommandSubscribe; @@ -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. + Throwable cause = ex.getCause() != null ? ex.getCause() : ex; + log.warn().attr("consumerName", ref.consumerName()).exceptionMessage(cause) + .log("ScalableTopicUnsubscribe failed"); + getCommandSender().sendErrorResponse(requestId, ServerError.UnknownError, + cause.getMessage()); + return; + } + // Removed only on success; a channelInactive racing the unregister just + // re-reports an already-removed session, which the coordinator ignores. + scalableConsumerRegistrations.remove(consumerId, ref); + getCommandSender().sendSuccessResponse(requestId); + }, ctx.executor()); + } + @Override protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata partitionMetadataParam) { checkArgument(state == State.Connected); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java index 3e5081e3985d2..c0366b0348964 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java +++ b/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 + * period covers that case. + */ + public CompletableFuture unregisterConsumer(TopicName topic, String subscription, + String consumerName) { + CompletableFuture future = controllers.get(topic.toString()); + if (future == null) { + return CompletableFuture.completedFuture(null); + } + return future.thenCompose(c -> c.unregisterConsumer(subscription, consumerName)); + } + // --- Internal helpers --- private CompletableFuture createUnderlyingSegmentTopic(TopicName parentTopic, SegmentInfo segment) { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java index 9e52890b7bda7..c05499f014821 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java @@ -444,21 +444,6 @@ public void testConsumerRejoiningAfterLeaveDoesNotWedgeRelease() throws Exceptio // watermark) must not leak into the last flip's drain: with everything acked and nothing // outstanding, a stale delivered-vs-acked pair would make the drain wait forever and the // rejoining consumer could never attach. - // A consumer's clean close is a disconnect to the controller, which holds its session for - // the grace period before rebalancing. Shrink it (read at coordinator creation, i.e. on - // this subscription's first register) so B1's departure hands the segment back promptly. - int defaultGrace = getPulsar().getConfiguration() - .getScalableTopicConsumerSessionGracePeriodSeconds(); - getPulsar().getConfiguration().setScalableTopicConsumerSessionGracePeriodSeconds(1); - try { - runRejoinAfterLeaveScenario(); - } finally { - getPulsar().getConfiguration() - .setScalableTopicConsumerSessionGracePeriodSeconds(defaultGrace); - } - } - - private void runRejoinAfterLeaveScenario() throws Exception { String topic = newScalableTopic(1); admin.scalableTopics().setAutoScalePolicy(topic, AutoScalePolicyOverride.builder().enabled(false).build()); @@ -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 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 b1 = v5Client.newStreamConsumer(Schema.string()) .topic(topic) .subscriptionName(subscription) .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST) @@ -515,9 +500,9 @@ private void runRejoinAfterLeaveScenario() throws Exception { tb1.join(); assertFalse(b1Got.isEmpty(), "consumer B1 received nothing — the segment did not fan out"); b1.close(); - b1Client.close(); - // Wait until the controller has handed the whole segment back to A and A completed the - // flip back to Exclusive — the stale-watermark state only matters once that is done. + // The clean leave unregisters immediately (no grace wait): the controller hands the + // whole segment back to A, which flips back to Exclusive. Only once that is done does + // the stale-watermark state matter for B2's rejoin. String segmentTopic = admin.scalableTopics().getStats(topic) .getSegments().values().iterator().next().name(); Awaitility.await().atMost(Duration.ofSeconds(15)).untilAsserted(() -> { diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java index 1597a7b083c68..745f9b6c3a497 100644 --- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java +++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java @@ -83,6 +83,8 @@ final class ScalableConsumerClient implements ScalableConsumerSession, AutoClose private volatile AssignmentChangeListener listener; private volatile ClientCnx cnx; private volatile boolean closed = false; + /** The in-flight (or last) subscribe attempt; close() awaits it before unsubscribing. */ + private volatile CompletableFuture lastSubscribeResult; ScalableConsumerClient(PulsarClientImpl v4Client, TopicName topicName, @@ -136,6 +138,11 @@ CompletableFuture> start() { */ private CompletableFuture connectAndSubscribe() { CompletableFuture result = new CompletableFuture<>(); + // Published before the closed-flag check below: close() chains its clean unsubscribe on + // this attempt, so either the attempt observes closed and never subscribes, or close() + // observes the attempt and waits for its outcome — never an unsubscribe that overtakes + // the in-flight registration (which would no-op and leave a ghost registration behind). + lastSubscribeResult = result; DagWatchClient watch = new DagWatchClient(v4Client, topicName); watch.start() @@ -374,11 +381,47 @@ public void close() { return; } closed = true; + CompletableFuture 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(); + c.getPendingRequests().put(requestId, responseFuture); + c.ctx().writeAndFlush(Commands.newScalableTopicUnsubscribe(requestId, consumerId)) + .addListener(writeFuture -> { + if (!writeFuture.isSuccess()) { + c.getPendingRequests().remove(requestId); + } + }); + responseFuture.exceptionally(ex -> { + log.debug().exceptionMessage(ex) + .log("Clean unsubscribe failed; relying on the grace period"); + return null; + }); + } catch (Exception e) { + log.debug().exceptionMessage(e) + .log("Clean unsubscribe failed; relying on the grace period"); } } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java index 575664e8b0930..5fff50798780b 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java @@ -1804,6 +1804,19 @@ public static ByteBuf newScalableTopicSubscribe(long requestId, String topic, return serializeWithSize(cmd); } + /** + * Client -> Broker: a scalable consumer is cleanly leaving its subscription; the + * controller unregisters it and rebalances immediately instead of waiting out the + * disconnect grace period. Acknowledged with {@code CommandSuccess}. + */ + public static ByteBuf newScalableTopicUnsubscribe(long requestId, long consumerId) { + BaseCommand cmd = localCmd(Type.SCALABLE_TOPIC_UNSUBSCRIBE); + cmd.setScalableTopicUnsubscribe() + .setRequestId(requestId) + .setConsumerId(consumerId); + return serializeWithSize(cmd); + } + /** * Broker -> Client: response to a scalable-topic subscribe request. On success the * caller must populate the nested {@link ScalableConsumerAssignment} via diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java index 9a412c1def16e..7cf04d830d1c6 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java @@ -77,6 +77,7 @@ import org.apache.pulsar.common.api.proto.CommandScalableTopicLookup; import org.apache.pulsar.common.api.proto.CommandScalableTopicSubscribe; import org.apache.pulsar.common.api.proto.CommandScalableTopicSubscribeResponse; +import org.apache.pulsar.common.api.proto.CommandScalableTopicUnsubscribe; import org.apache.pulsar.common.api.proto.CommandScalableTopicUpdate; import org.apache.pulsar.common.api.proto.CommandSeek; import org.apache.pulsar.common.api.proto.CommandSend; @@ -514,6 +515,11 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception handleCommandScalableTopicAssignmentUpdate(cmd.getScalableTopicAssignmentUpdate()); break; + case SCALABLE_TOPIC_UNSUBSCRIBE: + checkArgument(cmd.hasScalableTopicUnsubscribe()); + handleCommandScalableTopicUnsubscribe(cmd.getScalableTopicUnsubscribe()); + break; + case WATCH_SCALABLE_TOPICS: checkArgument(cmd.hasWatchScalableTopics()); handleCommandWatchScalableTopics(cmd.getWatchScalableTopics()); @@ -837,6 +843,11 @@ protected void handleCommandScalableTopicAssignmentUpdate( throw new UnsupportedOperationException(); } + protected void handleCommandScalableTopicUnsubscribe( + CommandScalableTopicUnsubscribe commandScalableTopicUnsubscribe) { + throw new UnsupportedOperationException(); + } + protected void handleCommandWatchScalableTopics( org.apache.pulsar.common.api.proto.CommandWatchScalableTopics commandWatchScalableTopics) { throw new UnsupportedOperationException(); diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto index 5d30972d93791..7d2d6845636f3 100644 --- a/pulsar-common/src/main/proto/PulsarApi.proto +++ b/pulsar-common/src/main/proto/PulsarApi.proto @@ -995,6 +995,16 @@ message CommandScalableTopicAssignmentUpdate { required ScalableConsumerAssignment assignment = 2; } +// Client -> Broker: a scalable consumer is cleanly leaving its subscription. The controller +// deletes the registration and rebalances the group immediately, instead of holding the +// session for the disconnect grace period (which remains the fallback for unclean +// departures). Acknowledged with CommandSuccess; idempotent — an unknown consumer_id (e.g. +// already swept by a disconnect) still succeeds. +message CommandScalableTopicUnsubscribe { + required uint64 request_id = 1; + required uint64 consumer_id = 2; +} + // Multi-topic consumer watcher: subscribes to the union of scalable topics in a // namespace that match a (possibly empty) set of property filters. The broker keeps // pushing updates as topics enter or leave the matching set. See @@ -1356,6 +1366,8 @@ message BaseCommand { WATCH_TC_ASSIGNMENTS = 79; WATCH_TC_ASSIGNMENTS_UPDATE = 80; WATCH_TC_ASSIGNMENTS_CLOSE = 81; + + SCALABLE_TOPIC_UNSUBSCRIBE = 82; } @@ -1455,4 +1467,6 @@ message BaseCommand { optional CommandWatchTcAssignments watchTcAssignments = 79; optional CommandWatchTcAssignmentsUpdate watchTcAssignmentsUpdate = 80; optional CommandWatchTcAssignmentsClose watchTcAssignmentsClose = 81; + + optional CommandScalableTopicUnsubscribe scalableTopicUnsubscribe = 82; } From 7ac6b40b153d6cf8844a19c2c7b1d11e59d1c7da Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Mon, 31 Aug 2026 16:12:05 -0700 Subject: [PATCH 2/2] [fix][client] Scalable clean leave: harden unregister against the review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses lhotari's review on the PR: 1. Failed metadata delete no longer wedges the group: the coordinator's unregisterConsumer now deletes the persisted registration first and only on success removes the in-memory session, cancels its grace timer, and rebalances. A failure changes nothing — the session stays registered and connected, so the channelInactive → grace fallback the ServerCnx error path relies on actually works, and a retried unsubscribe retries the deletion instead of short-circuiting on an already-removed session. 2. close() can no longer be raced by overlapping reconnect attempts: a connection drop used to schedule two reconnects (failed pending request + connectionClosed), and close() chained its unsubscribe on a single last-attempt slot that either could overwrite — letting the older attempt register after the unsubscribe. Reconnect scheduling is now collapsed to one pending attempt, and close() chains on every in-flight subscribe attempt. 3. close() deregisters the local session immediately and unconditionally again: deferring removeScalableConsumerSession behind the subscribe future meant an unanswered subscribe retained the closed session (and pinned the pooled connection non-idle) for the connection's lifetime. Only the wire command waits for the in-flight attempts now. 4. Both scalable session requests (subscribe and unsubscribe) go through a new ClientCnx.sendScalableSessionRequest helper with the standard bookkeeping — request-timeout tracking and write-failure completion — instead of hand-rolled pendingRequests entries that never timed out and, on write failure, left the future forever pending. 5. Direct coverage for the command: a Commands round-trip test, a ServerCnx test for the idempotency contract and the error path (error response keeps the registration ref, a retry re-runs the unregister, success removes the ref), and a subscribe/close churn e2e asserting the group converges back to a sole Exclusive owner under the default grace period. The rejoin e2e's leaver is now tracked for cleanup so an early assertion failure cannot leak it onto the shared client. 6. ScalableTopicService.unregisterConsumer javadoc: the no-op branch is reached by deleted topics / shutdown / never-registered consumers — not by moved leadership, which keeps its controller entry and fails via checkLeader() into the ref-preserving error path. Assisted-by: Claude Code (Fable 5) --- .../scalable/ScalableTopicService.java | 6 +- .../scalable/SubscriptionCoordinator.java | 16 ++- .../pulsar/broker/service/ServerCnxTest.java | 59 +++++++++ .../api/v5/V5EntryBucketDispatchTest.java | 44 ++++++- .../impl/v5/ScalableConsumerClient.java | 116 ++++++++++-------- .../apache/pulsar/client/impl/ClientCnx.java | 10 ++ .../protocol/CommandsScalableTopicTest.java | 9 ++ 7 files changed, 201 insertions(+), 59 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java index c0366b0348964..a3af75f7b63a0 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java @@ -319,8 +319,10 @@ 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 - * period covers that case. + * no controller entry exists here — which happens only for a consumer that never + * registered on this broker, a deleted topic, or a shutting-down service. (A deposed + * leader keeps its entry and fails via {@code checkLeader()}, taking the error path + * instead, which preserves the caller's registration ref and grace fallback.) */ public CompletableFuture unregisterConsumer(TopicName topic, String subscription, String consumerName) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinator.java index 22a8ccea0f511..a2c9a5acd1bb6 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinator.java @@ -214,19 +214,25 @@ public synchronized CompletableFuture> } /** - * 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> 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); + if (removed != null) { + removed.cancelGraceTimer(); + } if (sessions.isEmpty()) { segmentAssignments.clear(); return Map.of(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java index 7de4c198f30df..43a4a755566cd 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java @@ -1591,6 +1591,65 @@ public void testScalableTopicCommandsRequireTopicAuthorization() throws Exceptio channel.finish(); } + /** + * PIP-486 clean leave: unsubscribe for an unknown consumer id is an idempotent success; a + * failed unregister answers an error and keeps the per-connection registration ref (so the + * grace fallback and a retry both still work); a successful retry then removes it, and a + * repeat unsubscribe is again an idempotent success. + */ + @Test(timeOut = 30000) + public void testScalableTopicUnsubscribeIdempotencyAndErrorPath() throws Exception { + var scalableTopicService = + mock(org.apache.pulsar.broker.service.scalable.ScalableTopicService.class); + when(brokerService.getScalableTopicService()).thenReturn(scalableTopicService); + + resetChannel(); + ByteBuf connect = Commands.newConnect("none", "", null); + channel.writeInbound(connect); + assertTrue(getResponse() instanceof CommandConnected); + + // Unknown consumer id: idempotent success. + channel.writeInbound(Commands.newScalableTopicUnsubscribe(300L, 999L)); + Object response = getResponse(); + assertTrue(response instanceof CommandSuccess, String.valueOf(response)); + assertEquals(((CommandSuccess) response).getRequestId(), 300L); + + // Register a consumer so the connection records its registration ref. + when(scalableTopicService.registerConsumer(any(), anyString(), anyString(), anyLong(), + any(), any())).thenReturn(CompletableFuture.completedFuture( + new org.apache.pulsar.broker.service.scalable.ConsumerAssignment( + 1L, Collections.emptyList()))); + channel.writeInbound(Commands.newScalableTopicSubscribe(301L, + "persistent://public/default/scalable-unsub", "sub", "c1", 5L, + ScalableConsumerType.STREAM)); + assertTrue(getResponse() instanceof CommandScalableTopicSubscribeResponse); + + // Failing unregister: error response, ref retained. + when(scalableTopicService.unregisterConsumer(any(), anyString(), anyString())) + .thenReturn(CompletableFuture.failedFuture(new RuntimeException("store down"))); + channel.writeInbound(Commands.newScalableTopicUnsubscribe(302L, 5L)); + response = getResponse(); + assertTrue(response instanceof CommandError, String.valueOf(response)); + assertEquals(((CommandError) response).getRequestId(), 302L); + + // Retry after the failure actually retries the unregister (the ref survived). + when(scalableTopicService.unregisterConsumer(any(), anyString(), anyString())) + .thenReturn(CompletableFuture.completedFuture(null)); + channel.writeInbound(Commands.newScalableTopicUnsubscribe(303L, 5L)); + response = getResponse(); + assertTrue(response instanceof CommandSuccess, String.valueOf(response)); + assertEquals(((CommandSuccess) response).getRequestId(), 303L); + verify(scalableTopicService, times(2)).unregisterConsumer(any(), anyString(), anyString()); + + // The ref is gone now: one more unsubscribe is an idempotent success with no new call. + channel.writeInbound(Commands.newScalableTopicUnsubscribe(304L, 5L)); + response = getResponse(); + assertTrue(response instanceof CommandSuccess, String.valueOf(response)); + verify(scalableTopicService, times(2)).unregisterConsumer(any(), anyString(), anyString()); + + channel.finish(); + } + @Test public void testRefreshOriginalPrincipalWithAuthDataForwardedFromProxy() throws Exception { AuthenticationService authenticationService = mock(AuthenticationService.class); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java index c05499f014821..c764321657529 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java @@ -487,11 +487,11 @@ public void testConsumerRejoiningAfterLeaveDoesNotWedgeRelease() throws Exceptio // 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 b1 = v5Client.newStreamConsumer(Schema.string()) + StreamConsumer b1 = track(v5Client.newStreamConsumer(Schema.string()) .topic(topic) .subscriptionName(subscription) .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST) - .subscribeAsync().get(30, TimeUnit.SECONDS); + .subscribeAsync().get(30, TimeUnit.SECONDS)); sendPhase(producer, keys, sent, perKey, perKey * 2); Map> b1Got = new ConcurrentHashMap<>(); Thread ta1 = drainOrdered(a, aGot); @@ -544,6 +544,46 @@ public void testConsumerRejoiningAfterLeaveDoesNotWedgeRelease() throws Exceptio } } + @Test + public void testRapidSubscribeCloseChurnLeavesGroupClean() throws Exception { + // PIP-486 clean leave under churn: consumers joining and closing in quick succession — + // including closes racing the registration and rebalance machinery — must leave no + // ghost group members behind. A single ghost would make the controller keep the + // segment fanned out (Key_Shared) instead of returning it to the survivor Exclusive. + String topic = newScalableTopic(1); + admin.scalableTopics().setAutoScalePolicy(topic, + AutoScalePolicyOverride.builder().enabled(false).build()); + String subscription = "leave-churn"; + + @Cleanup + StreamConsumer survivor = v5Client.newStreamConsumer(Schema.string()) + .topic(topic) + .subscriptionName(subscription) + .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST) + .subscribe(); + for (int i = 0; i < 5; i++) { + v5Client.newStreamConsumer(Schema.string()) + .topic(topic) + .subscriptionName(subscription) + .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST) + .subscribeAsync().get(30, TimeUnit.SECONDS) + .close(); + } + + // Every churned consumer unregistered cleanly: the survivor converges back to sole + // Exclusive ownership. With the default 60s grace, any missed unregistration would + // hold the group fanned out well past this window. + String segmentTopic = admin.scalableTopics().getStats(topic) + .getSegments().values().iterator().next().name(); + Awaitility.await().atMost(Duration.ofSeconds(20)).untilAsserted(() -> { + var sub = getTopicReference(segmentTopic).orElseThrow().getSubscription(subscription); + assertNotNull(sub, "segment subscription missing"); + assertEquals(sub.getType(), CommandSubscribe.SubType.Exclusive, + "a churned consumer left a ghost registration behind"); + assertEquals(sub.getConsumers().size(), 1); + }); + } + private void sendPhase(Producer producer, List keys, Map> sent, int from, int to) throws Exception { for (int i = from; i < to; i++) { diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java index 745f9b6c3a497..af2c7242b3b21 100644 --- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java +++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java @@ -25,15 +25,17 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.impl.ClientCnx; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.client.impl.ScalableConsumerSession; import org.apache.pulsar.client.impl.v5.SegmentRouter.ActiveSegment; -import org.apache.pulsar.client.util.TimedCompletableFuture; import org.apache.pulsar.common.api.proto.ScalableAssignedSegment; import org.apache.pulsar.common.api.proto.ScalableConsumerAssignment; import org.apache.pulsar.common.api.proto.ScalableConsumerType; @@ -83,8 +85,15 @@ final class ScalableConsumerClient implements ScalableConsumerSession, AutoClose private volatile AssignmentChangeListener listener; private volatile ClientCnx cnx; private volatile boolean closed = false; - /** The in-flight (or last) subscribe attempt; close() awaits it before unsubscribing. */ - private volatile CompletableFuture lastSubscribeResult; + /** + * Every subscribe attempt still in flight; close() awaits all of them before + * unsubscribing (a single last-attempt slot can be overwritten when two reconnect + * attempts overlap, letting the older one register after the unsubscribe was sent). + */ + private final Set> inFlightSubscribes = + ConcurrentHashMap.newKeySet(); + /** Collapses concurrent reconnect triggers (request failure + connectionClosed) into one. */ + private final AtomicBoolean reconnectPending = new AtomicBoolean(false); ScalableConsumerClient(PulsarClientImpl v4Client, TopicName topicName, @@ -138,11 +147,14 @@ CompletableFuture> start() { */ private CompletableFuture connectAndSubscribe() { CompletableFuture result = new CompletableFuture<>(); - // Published before the closed-flag check below: close() chains its clean unsubscribe on - // this attempt, so either the attempt observes closed and never subscribes, or close() - // observes the attempt and waits for its outcome — never an unsubscribe that overtakes - // the in-flight registration (which would no-op and leave a ghost registration behind). - lastSubscribeResult = result; + // Registered before the closed-flag check below: close() chains its clean unsubscribe + // on every attempt in this set, so either an attempt observes closed and never + // subscribes, or close() observes the attempt and waits for its outcome — never an + // unsubscribe that overtakes an in-flight registration (which would no-op and leave a + // ghost registration behind). Settled attempts drop out; their registrations, if any, + // already exist and are covered by the unsubscribe that follows. + inFlightSubscribes.add(result); + result.whenComplete((__, ___) -> inFlightSubscribes.remove(result)); DagWatchClient watch = new DagWatchClient(v4Client, topicName); watch.start() @@ -193,32 +205,26 @@ private CompletableFuture connectAndSubscribe() { cnx.registerScalableConsumerSession(consumerId, this); long requestId = v4Client.newRequestId(); - var responseFuture = new TimedCompletableFuture(); - cnx.getPendingRequests().put(requestId, responseFuture); - - cnx.ctx().writeAndFlush(Commands.newScalableTopicSubscribe( - requestId, - topicName.toString(), - subscription, - consumerName, - consumerId, - consumerType)) - .addListener(writeFuture -> { - if (!writeFuture.isSuccess()) { - cnx.getPendingRequests().remove(requestId); + cnx.sendScalableSessionRequest( + Commands.newScalableTopicSubscribe( + requestId, + topicName.toString(), + subscription, + consumerName, + consumerId, + consumerType), + requestId) + .whenComplete((assignment, ex) -> { + if (ex != null) { + // Write failure, error response, or request timeout: the + // broker holds no push route for us, so drop the local + // session registration too (retries re-register). cnx.removeScalableConsumerSession(consumerId); - result.completeExceptionally( - new PulsarClientException(writeFuture.cause())); + result.completeExceptionally(ex); + } else { + result.complete(assignment); } }); - - responseFuture.whenComplete((assignment, ex) -> { - if (ex != null) { - result.completeExceptionally(ex); - } else { - result.complete(assignment); - } - }); }) .exceptionally(ex -> { result.completeExceptionally(ex); @@ -287,6 +293,12 @@ private void scheduleReconnect() { if (closed) { return; } + // A connection drop triggers this twice (the failed pending request and the + // connectionClosed callback); collapse to a single scheduled attempt so two + // overlapping connectAndSubscribe() calls never run. + if (!reconnectPending.compareAndSet(false, true)) { + return; + } long delayMs = reconnectBackoff.next().toMillis(); log.info().attr("delayMs", delayMs).log("Scheduling reconnect"); v4Client.timer().newTimeout(timeout -> reconnect(), @@ -294,6 +306,7 @@ private void scheduleReconnect() { } private void reconnect() { + reconnectPending.set(false); if (closed) { return; } @@ -381,21 +394,30 @@ public void close() { return; } closed = true; - CompletableFuture pending = lastSubscribeResult; - if (pending == null) { - // Never attempted to subscribe: nothing registered anywhere. - return; + // Deregister the local session immediately and unconditionally (as before the clean + // leave existed): the subscribe response routes through pendingRequests, not through + // the session registry, so nothing here needs to wait — and waiting on a subscribe + // that is never answered would retain the closed session (and pin the pooled + // connection non-idle) for the connection's lifetime. + ClientCnx c = cnx; + if (c != null) { + c.removeScalableConsumerSession(consumerId); } // 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 + // Chained on every in-flight subscribe attempt so the unsubscribe can never overtake a // 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 + // when a 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()); + CompletableFuture[] pending = inFlightSubscribes.toArray(CompletableFuture[]::new); + if (pending.length == 0) { + sendUnsubscribe(); + } else { + CompletableFuture.allOf(pending).whenComplete((__, ___) -> sendUnsubscribe()); + } } private void sendUnsubscribe() { @@ -406,19 +428,13 @@ private void sendUnsubscribe() { c.removeScalableConsumerSession(consumerId); try { long requestId = v4Client.newRequestId(); - var responseFuture = new TimedCompletableFuture(); - c.getPendingRequests().put(requestId, responseFuture); - c.ctx().writeAndFlush(Commands.newScalableTopicUnsubscribe(requestId, consumerId)) - .addListener(writeFuture -> { - if (!writeFuture.isSuccess()) { - c.getPendingRequests().remove(requestId); - } + c.sendScalableSessionRequest( + Commands.newScalableTopicUnsubscribe(requestId, consumerId), requestId) + .exceptionally(ex -> { + log.debug().exceptionMessage(ex) + .log("Clean unsubscribe failed; relying on the grace period"); + return null; }); - responseFuture.exceptionally(ex -> { - log.debug().exceptionMessage(ex) - .log("Clean unsubscribe failed; relying on the grace period"); - return null; - }); } catch (Exception e) { log.debug().exceptionMessage(e) .log("Clean unsubscribe failed; relying on the grace period"); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java index 719bd728ff500..4bd1396e032fe 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java @@ -1468,6 +1468,16 @@ CompletableFuture sendRequestWithId(ByteBuf cmd, long requestI return sendRequestAndHandleTimeout(cmd, requestId, RequestType.Command, true); } + /** + * Send a scalable-topic session request (PIP-468/486) with the standard request + * bookkeeping — pending-request registration, request-timeout tracking, and + * write-failure completion — instead of hand-rolling it at the call site. The matching + * response handler completes the returned future by request id. + */ + public CompletableFuture sendScalableSessionRequest(ByteBuf requestMessage, long requestId) { + return sendRequestAndHandleTimeout(requestMessage, requestId, RequestType.Command, true); + } + private void sendRequestAndHandleTimeout(ByteBuf requestMessage, long requestId, RequestType requestType, boolean flush, TimedCompletableFuture future) { diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/CommandsScalableTopicTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/CommandsScalableTopicTest.java index a5d028889eea4..af9db0da4ab4d 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/CommandsScalableTopicTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/protocol/CommandsScalableTopicTest.java @@ -169,6 +169,15 @@ public void testNewScalableTopicError() { "Scalable topic not found: topic://t/n/x"); } + @Test + public void testNewScalableTopicUnsubscribe() { + BaseCommand cmd = parseFrame(Commands.newScalableTopicUnsubscribe(42L, 7L)); + assertEquals(cmd.getType(), BaseCommand.Type.SCALABLE_TOPIC_UNSUBSCRIBE); + assertTrue(cmd.hasScalableTopicUnsubscribe()); + assertEquals(cmd.getScalableTopicUnsubscribe().getRequestId(), 42L); + assertEquals(cmd.getScalableTopicUnsubscribe().getConsumerId(), 7L); + } + @Test public void testNewScalableTopicSubscribeResponseSuccess() { ScalableConsumerAssignment assignment = new ScalableConsumerAssignment().setLayoutEpoch(3L);