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 42988cabfd96f..63155339c296b 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..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 @@ -316,6 +316,23 @@ 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 + * 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) { + 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/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 9e52890b7bda7..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 @@ -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,16 +482,16 @@ 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 = 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); @@ -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(() -> { @@ -559,6 +544,46 @@ private void runRejoinAfterLeaveScenario() throws Exception { } } + @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 1597a7b083c68..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,6 +85,15 @@ final class ScalableConsumerClient implements ScalableConsumerSession, AutoClose private volatile AssignmentChangeListener listener; private volatile ClientCnx cnx; private volatile boolean closed = false; + /** + * 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, @@ -136,6 +147,14 @@ CompletableFuture> start() { */ private CompletableFuture connectAndSubscribe() { CompletableFuture result = new CompletableFuture<>(); + // 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() @@ -186,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); @@ -280,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(), @@ -287,6 +306,7 @@ private void scheduleReconnect() { } private void reconnect() { + reconnectPending.set(false); if (closed) { return; } @@ -374,11 +394,50 @@ public void close() { return; } closed = true; + // 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); - // No close command for now — broker reaps registrations via grace timer on - // disconnect. A future refactor can add an explicit unsubscribe. + } + // 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 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 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. + CompletableFuture[] pending = inFlightSubscribes.toArray(CompletableFuture[]::new); + if (pending.length == 0) { + sendUnsubscribe(); + } else { + CompletableFuture.allOf(pending).whenComplete((__, ___) -> sendUnsubscribe()); + } + } + + private void sendUnsubscribe() { + ClientCnx c = cnx; + if (c == null) { + return; + } + c.removeScalableConsumerSession(consumerId); + try { + long requestId = v4Client.newRequestId(); + c.sendScalableSessionRequest( + Commands.newScalableTopicUnsubscribe(requestId, consumerId), requestId) + .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/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; } 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);