You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/12 19:18:49 UTC
(pulsar) 24/27: [fix][broker] Answer a scalable consumer's subscribe in order with its assignment updates (#26857)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 845589bbef731b1beac99aaaa96f26d53e7ac861
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Mon Oct 12 12:15:08 2026 -0700
[fix][broker] Answer a scalable consumer's subscribe in order with its assignment updates (#26857)
(cherry picked from commit 45104a51d09c0ac0057765aaf206e97f9497fcfe)
---
.../apache/pulsar/broker/service/ServerCnx.java | 90 ++++++++--
.../apache/pulsar/broker/service/TransportCnx.java | 8 +
.../broker/service/scalable/ConsumerSession.java | 13 +-
.../service/scalable/SubscriptionCoordinator.java | 4 +
.../pulsar/broker/service/ServerCnxTest.java | 107 ++++++++++++
.../service/scalable/ConsumerSessionTest.java | 22 +--
.../impl/v5/V5StaleSubscribeResponseTest.java | 182 +++++++++++++++++++++
7 files changed, 382 insertions(+), 44 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 5a0c9fcb248f..c851333a7286 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
@@ -64,6 +64,7 @@ import java.util.concurrent.CompletionStage;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BooleanSupplier;
import java.util.stream.Collectors;
import javax.naming.AuthenticationException;
@@ -101,8 +102,10 @@ import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionNotFo
import org.apache.pulsar.broker.service.BrokerServiceException.TopicNotFoundException;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.service.scalable.ConsumerAssignment;
import org.apache.pulsar.broker.service.scalable.ConsumerSession;
import org.apache.pulsar.broker.service.scalable.DagWatchSession;
+import org.apache.pulsar.broker.service.scalable.ScalableTopicService;
import org.apache.pulsar.broker.service.scalable.ScalableTopicsWatcherSession;
import org.apache.pulsar.broker.service.schema.SchemaRegistryService;
import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
@@ -164,6 +167,7 @@ import org.apache.pulsar.common.api.proto.MessageIdData;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.api.proto.ProducerAccessMode;
import org.apache.pulsar.common.api.proto.ProtocolVersion;
+import org.apache.pulsar.common.api.proto.ScalableConsumerAssignment;
import org.apache.pulsar.common.api.proto.ScalableConsumerType;
import org.apache.pulsar.common.api.proto.Schema;
import org.apache.pulsar.common.api.proto.ServerError;
@@ -1153,11 +1157,18 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
private final java.util.concurrent.ConcurrentHashMap<Long, ScalableConsumerRegistrationRef>
scalableConsumerRegistrations = new java.util.concurrent.ConcurrentHashMap<>();
+ /**
+ * A scalable-topic consumer registration made on this connection. {@code answered} records
+ * whether the response to its subscribe, {@code requestId}, was written; see
+ * {@link #sendScalableConsumerAssignment}.
+ */
private record ScalableConsumerRegistrationRef(
TopicName topicName,
String subscription,
String consumerName,
- CompletableFuture<?> registration) {}
+ CompletableFuture<?> registration,
+ long requestId,
+ AtomicBoolean answered) {}
@Override
protected void handleCommandScalableTopicSubscribe(
@@ -1212,31 +1223,45 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
ServerError.AuthorizationError, msg);
return;
}
- // Record the registration BEFORE it resolves, carrying its future: an
+ // Record the registration BEFORE starting it, carrying its future: an
// unsubscribe (or the channelInactive sweep) arriving mid-registration
// chains behind it instead of silently missing it. The client's subscribe
// can time out while the broker-side registration is still in flight, so
- // this ordering must not depend on how long the client was able to wait.
- var registration = scalableTopicService.registerConsumer(topicName,
- subscription, consumerName, consumerId, consumerType, this);
- var ref = new ScalableConsumerRegistrationRef(
- topicName, subscription, consumerName, registration);
+ // this ordering must not depend on how long the client was able to wait;
+ // and the subscribe can be answered before registerScalableConsumer even
+ // returns (see sendScalableConsumerAssignment).
+ var registration = new CompletableFuture<ConsumerAssignment>();
+ var ref = new ScalableConsumerRegistrationRef(topicName, subscription,
+ consumerName, registration, requestId, new AtomicBoolean());
scalableConsumerRegistrations.put(consumerId, ref);
+ FutureUtil.completeAfter(registration, FutureUtil.supplySafely(() ->
+ registerScalableConsumer(scalableTopicService, topicName, subscription,
+ consumerName, consumerId, consumerType)));
registration.whenCompleteAsync((assignment, ex) -> {
+ // Normally answered already, by the push of the consumer's assignment
+ // the coordinator makes while registering it.
+ boolean unanswered = ref.answered().compareAndSet(false, true);
if (ex != null) {
Throwable cause = ex.getCause() != null ? ex.getCause() : ex;
log.warn().attr("topic", topicName).attr("subscription", subscription)
- .attr("consumerName", consumerName).exception(cause)
- .log("ScalableTopicSubscribe failed");
- // Nothing was registered: drop the ref so unsubscribes and the
- // disconnect sweep have nothing to report for it.
- scalableConsumerRegistrations.remove(consumerId, ref);
- getCommandSender().sendScalableTopicSubscribeError(requestId,
- ServerError.UnknownError, cause.getMessage());
+ .attr("consumerName", consumerName).attr("answered", !unanswered)
+ .exception(cause).log("ScalableTopicSubscribe failed");
+ if (unanswered) {
+ // Nothing was registered: drop the ref so unsubscribes and the
+ // disconnect sweep have nothing to report for it.
+ scalableConsumerRegistrations.remove(consumerId, ref);
+ getCommandSender().sendScalableTopicSubscribeError(requestId,
+ ServerError.UnknownError, cause.getMessage());
+ }
+ // Otherwise the coordinator registered the consumer and answered the
+ // subscribe before the failure: keep the ref, so the consumer's
+ // unsubscribe and the disconnect sweep still reach the coordinator.
return;
}
- getCommandSender().sendScalableTopicSubscribeResponse(requestId,
- ConsumerSession.toProto(assignment));
+ if (unanswered) {
+ getCommandSender().sendScalableTopicSubscribeResponse(requestId,
+ ConsumerSession.toProto(assignment));
+ }
}, ctx.executor());
})
.exceptionally(ex -> {
@@ -1249,6 +1274,39 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
});
}
+ /**
+ * Register a scalable-topic consumer with the controller leader. Tests override it to hold
+ * the registration, as this connection sees it, while other members of the subscription
+ * register.
+ */
+ @VisibleForTesting
+ protected CompletableFuture<ConsumerAssignment> registerScalableConsumer(
+ ScalableTopicService scalableTopicService, TopicName topicName, String subscription,
+ String consumerName, long consumerId, ScalableConsumerType consumerType) {
+ return scalableTopicService.registerConsumer(topicName, subscription, consumerName,
+ consumerId, consumerType, this);
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * <p>The coordinator pushes a registering consumer its assignment while holding its monitor,
+ * as part of the registration (of a new session and of a reconnect), so answering the
+ * subscribe with that push orders the response with every later push. Answering only when the
+ * registration future completes would let a rebalance pushed in between, by another member
+ * registering, reach the client first: both carry the same layout epoch, so the client would
+ * apply the older response last and keep it until the next rebalance.
+ */
+ @Override
+ public void sendScalableConsumerAssignment(long consumerId, ScalableConsumerAssignment assignment) {
+ ScalableConsumerRegistrationRef ref = scalableConsumerRegistrations.get(consumerId);
+ if (ref != null && ref.answered().compareAndSet(false, true)) {
+ getCommandSender().sendScalableTopicSubscribeResponse(ref.requestId(), assignment);
+ } else {
+ getCommandSender().sendScalableTopicAssignmentUpdate(consumerId, assignment);
+ }
+ }
+
@Override
protected void handleCommandScalableTopicUnsubscribe(
CommandScalableTopicUnsubscribe commandScalableTopicUnsubscribe) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java
index 2c0b247a94b7..53c071a5780f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java
@@ -26,6 +26,7 @@ import java.util.concurrent.CompletableFuture;
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData;
import org.apache.pulsar.common.api.proto.FeatureFlags;
+import org.apache.pulsar.common.api.proto.ScalableConsumerAssignment;
public interface TransportCnx {
@@ -98,4 +99,11 @@ public interface TransportCnx {
return getFeatures() != null && getFeatures().hasSupportsReplDedupByLidAndEid()
&& getFeatures().isSupportsReplDedupByLidAndEid();
}
+
+ /**
+ * Send a scalable-topic consumer an assignment pushed by its controller: as the response to the
+ * consumer's subscribe while that is unanswered on this connection, as an assignment update
+ * otherwise.
+ */
+ void sendScalableConsumerAssignment(long consumerId, ScalableConsumerAssignment assignment);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ConsumerSession.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ConsumerSession.java
index 99061daa73dc..71e62ef14ecf 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ConsumerSession.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ConsumerSession.java
@@ -168,9 +168,9 @@ public class ConsumerSession {
}
/**
- * Push an assignment update to this consumer, if currently connected. Builds a
- * {@code ScalableConsumerAssignment} proto and writes a
- * {@code CommandScalableTopicAssignmentUpdate} to the connection.
+ * Push an assignment to this consumer, if currently connected. The connection sends it as the
+ * response to the consumer's subscribe while that is unanswered, and as a
+ * {@code CommandScalableTopicAssignmentUpdate} otherwise.
*/
public void sendAssignmentUpdate(ConsumerAssignment assignment) {
TransportCnx localCnx = this.cnx;
@@ -179,12 +179,7 @@ public class ConsumerSession {
// reconnects (the coordinator re-pushes on attach).
return;
}
- var sender = localCnx.getCommandSender();
- if (sender == null) {
- // Connection is in the middle of being torn down; skip silently.
- return;
- }
- sender.sendScalableTopicAssignmentUpdate(consumerId, toProto(assignment));
+ localCnx.sendScalableConsumerAssignment(consumerId, toProto(assignment));
}
/**
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 73950aba78ec..bbf01f5eb859 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
@@ -189,6 +189,10 @@ public class SubscriptionCoordinator {
* is preserved and the new connection is attached; otherwise the registration is
* persisted and a rebalance is triggered.
*
+ * <p>Both paths push the registering consumer its assignment while holding the monitor. The
+ * connection answers the consumer's subscribe with that push, which orders the response with
+ * every later push (see {@code ServerCnx#sendScalableConsumerAssignment}).
+ *
* @return assignment map for all consumers (unchanged on reconnect, recomputed on fresh register)
*/
public synchronized CompletableFuture<Map<ConsumerSession, ConsumerAssignment>> registerConsumer(
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 de022c050dd4..ffe390458f41 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
@@ -60,6 +60,7 @@ import java.io.Closeable;
import java.io.IOException;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -77,6 +78,7 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@@ -120,6 +122,9 @@ import org.apache.pulsar.broker.namespace.TopicExistsInfo;
import org.apache.pulsar.broker.service.BrokerServiceException.ServiceUnitNotReadyException;
import org.apache.pulsar.broker.service.ServerCnx.State;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.service.scalable.ConsumerAssignment;
+import org.apache.pulsar.broker.service.scalable.ConsumerSession;
+import org.apache.pulsar.broker.service.scalable.ScalableTopicService;
import org.apache.pulsar.broker.service.schema.SchemaRegistryService;
import org.apache.pulsar.broker.service.utils.ClientChannelHelper;
import org.apache.pulsar.broker.testcontext.PulsarTestContext;
@@ -150,6 +155,7 @@ import org.apache.pulsar.common.api.proto.CommandLookupTopicResponse;
import org.apache.pulsar.common.api.proto.CommandPartitionedTopicMetadataResponse;
import org.apache.pulsar.common.api.proto.CommandPing;
import org.apache.pulsar.common.api.proto.CommandProducerSuccess;
+import org.apache.pulsar.common.api.proto.CommandScalableTopicAssignmentUpdate;
import org.apache.pulsar.common.api.proto.CommandScalableTopicSubscribeResponse;
import org.apache.pulsar.common.api.proto.CommandScalableTopicUpdate;
import org.apache.pulsar.common.api.proto.CommandSendError;
@@ -180,6 +186,7 @@ import org.apache.pulsar.common.protocol.PulsarHandler;
import org.apache.pulsar.common.protocol.schema.EmptyVersion;
import org.apache.pulsar.common.protocol.schema.SchemaData;
import org.apache.pulsar.common.protocol.schema.SchemaVersion;
+import org.apache.pulsar.common.scalable.HashRange;
import org.apache.pulsar.common.topics.TopicList;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap;
@@ -1723,6 +1730,106 @@ public class ServerCnxTest {
channel.finish();
}
+ /**
+ * A scalable subscribe is answered with the assignment the coordinator pushes the consumer while
+ * registering it, so a rebalance pushed before the registration completes (another member
+ * registering) follows the response instead of overtaking it. Both carry the same layout epoch:
+ * a client that got the older response last would keep it.
+ */
+ @Test(timeOut = 30000)
+ public void testScalableTopicSubscribeAnsweredInCoordinatorOrder() throws Exception {
+ ScalableTopicService scalableTopicService = mock(ScalableTopicService.class);
+ when(brokerService.getScalableTopicService()).thenReturn(scalableTopicService);
+ CompletableFuture<ConsumerAssignment> registration = new CompletableFuture<>();
+ AtomicReference<ConsumerSession> session = new AtomicReference<>();
+ // As the coordinator does: push the registering consumer its assignment, and complete the
+ // registration afterwards.
+ when(scalableTopicService.registerConsumer(any(), anyString(), anyString(), anyLong(), any(), any()))
+ .thenAnswer(invocation -> {
+ session.set(new ConsumerSession("c1", 7L, invocation.getArgument(5),
+ Duration.ofMinutes(1), mock(ScheduledExecutorService.class), () -> {}, log));
+ session.get().sendAssignmentUpdate(scalableAssignment(0L, 1L));
+ return registration;
+ });
+
+ resetChannel();
+ channel.writeInbound(Commands.newConnect("none", "", null));
+ assertTrue(getResponse() instanceof CommandConnected);
+
+ channel.writeInbound(Commands.newScalableTopicSubscribe(500L,
+ "persistent://public/default/scalable-order", "group", "c1", 7L,
+ ScalableConsumerType.CHECKPOINT));
+ Awaitility.await().until(() -> session.get() != null);
+ // A second member registers before this registration completes: c1 keeps only segment 0.
+ session.get().sendAssignmentUpdate(scalableAssignment(0L));
+ registration.complete(scalableAssignment(0L, 1L));
+
+ Object first = getResponse();
+ assertTrue(first instanceof CommandScalableTopicSubscribeResponse, String.valueOf(first));
+ assertEquals(((CommandScalableTopicSubscribeResponse) first).getRequestId(), 500L);
+ assertEquals(((CommandScalableTopicSubscribeResponse) first).getAssignment().getSegmentsCount(), 2);
+ Object second = getResponse();
+ assertTrue(second instanceof CommandScalableTopicAssignmentUpdate, String.valueOf(second));
+ assertEquals(((CommandScalableTopicAssignmentUpdate) second).getConsumerId(), 7L);
+ assertEquals(((CommandScalableTopicAssignmentUpdate) second).getAssignment().getSegmentsCount(), 1);
+ channel.runPendingTasks();
+ assertTrue(channel.outboundMessages().isEmpty(),
+ "the completed registration must not answer the subscribe again");
+
+ channel.finish();
+ }
+
+ /**
+ * A registration that fails after the coordinator's push answered the subscribe sends no error
+ * for the already answered request, and keeps the registration ref: the coordinator holds the
+ * session, so the consumer's unsubscribe must still reach it.
+ */
+ @Test(timeOut = 30000)
+ public void testScalableTopicSubscribeFailingAfterAnswerKeepsRegistration() throws Exception {
+ ScalableTopicService scalableTopicService = mock(ScalableTopicService.class);
+ when(brokerService.getScalableTopicService()).thenReturn(scalableTopicService);
+ when(scalableTopicService.registerConsumer(any(), anyString(), anyString(), anyLong(), any(), any()))
+ .thenAnswer(invocation -> {
+ new ConsumerSession("c1", 7L, invocation.getArgument(5), Duration.ofMinutes(1),
+ mock(ScheduledExecutorService.class), () -> {}, log)
+ .sendAssignmentUpdate(scalableAssignment(0L));
+ return CompletableFuture.failedFuture(new RuntimeException("failed after registering"));
+ });
+ when(scalableTopicService.unregisterConsumer(any(), anyString(), anyString(), anyLong()))
+ .thenReturn(CompletableFuture.completedFuture(null));
+
+ resetChannel();
+ channel.writeInbound(Commands.newConnect("none", "", null));
+ assertTrue(getResponse() instanceof CommandConnected);
+
+ channel.writeInbound(Commands.newScalableTopicSubscribe(600L,
+ "persistent://public/default/scalable-late-failure", "group", "c1", 7L,
+ ScalableConsumerType.CHECKPOINT));
+ Object response = getResponse();
+ assertTrue(response instanceof CommandScalableTopicSubscribeResponse, String.valueOf(response));
+ assertFalse(((CommandScalableTopicSubscribeResponse) response).hasError());
+ channel.runPendingTasks();
+ assertTrue(channel.outboundMessages().isEmpty(), "no error may follow the answered subscribe");
+
+ // The ref was kept: the unsubscribe reaches the coordinator instead of answering as a no-op.
+ channel.writeInbound(Commands.newScalableTopicUnsubscribe(601L, 7L));
+ Object unsubscribed = getResponse();
+ assertTrue(unsubscribed instanceof CommandSuccess, String.valueOf(unsubscribed));
+ verify(scalableTopicService).unregisterConsumer(any(), anyString(), anyString(), eq(7L));
+
+ channel.finish();
+ }
+
+ private static ConsumerAssignment scalableAssignment(long... segmentIds) {
+ List<ConsumerAssignment.AssignedSegment> segments = new ArrayList<>();
+ for (long segmentId : segmentIds) {
+ int start = (int) segmentId * 0x8000;
+ segments.add(new ConsumerAssignment.AssignedSegment(segmentId, HashRange.of(start, start + 0x7FFF),
+ "segment://public/default/scalable-order/" + segmentId, List.of()));
+ }
+ return new ConsumerAssignment(1L, segments);
+ }
+
@Test
public void testRefreshOriginalPrincipalWithAuthDataForwardedFromProxy() throws Exception {
AuthenticationService authenticationService = mock(AuthenticationService.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ConsumerSessionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ConsumerSessionTest.java
index f2fee9e4bf98..11dec746aaed 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ConsumerSessionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ConsumerSessionTest.java
@@ -38,7 +38,6 @@ import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
-import org.apache.pulsar.broker.service.PulsarCommandSender;
import org.apache.pulsar.broker.service.TransportCnx;
import org.apache.pulsar.common.api.proto.ScalableConsumerAssignment;
import org.apache.pulsar.common.scalable.HashRange;
@@ -212,18 +211,16 @@ public class ConsumerSessionTest {
// --- sendAssignmentUpdate ---
@Test
- public void testSendAssignmentUpdateWritesViaCommandSender() {
+ public void testSendAssignmentUpdateWritesToConnection() {
TestContext ctx = new TestContext();
TransportCnx cnx = mock(TransportCnx.class);
- PulsarCommandSender sender = mock(PulsarCommandSender.class);
- when(cnx.getCommandSender()).thenReturn(sender);
ConsumerSession session = ctx.session("c1", 42L, cnx);
session.sendAssignmentUpdate(buildAssignment(5L, 0L, 1L));
ArgumentCaptor<ScalableConsumerAssignment> captor =
ArgumentCaptor.forClass(ScalableConsumerAssignment.class);
- verify(sender).sendScalableTopicAssignmentUpdate(eq(42L), captor.capture());
+ verify(cnx).sendScalableConsumerAssignment(eq(42L), captor.capture());
ScalableConsumerAssignment proto = captor.getValue();
assertEquals(proto.getLayoutEpoch(), 5L);
assertEquals(proto.getSegmentsCount(), 2);
@@ -244,25 +241,12 @@ public class ConsumerSessionTest {
public void testSendAssignmentUpdateIsNoopAfterMarkDisconnected() {
TestContext ctx = new TestContext();
TransportCnx cnx = mock(TransportCnx.class);
- PulsarCommandSender sender = mock(PulsarCommandSender.class);
- when(cnx.getCommandSender()).thenReturn(sender);
ConsumerSession session = ctx.session("c1", 10L, cnx);
session.markDisconnected();
session.sendAssignmentUpdate(buildAssignment(0L, 0L));
- verify(sender, never()).sendScalableTopicAssignmentUpdate(anyLong(), any());
- }
-
- @Test
- public void testSendAssignmentUpdateIsNoopWhenCommandSenderNull() {
- TestContext ctx = new TestContext();
- TransportCnx cnx = mock(TransportCnx.class);
- when(cnx.getCommandSender()).thenReturn(null); // connection tearing down
- ConsumerSession session = ctx.session("c1", 10L, cnx);
-
- // no exception, no interaction
- session.sendAssignmentUpdate(buildAssignment(0L, 0L));
+ verify(cnx, never()).sendScalableConsumerAssignment(anyLong(), any());
}
// --- toProto ---
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5StaleSubscribeResponseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5StaleSubscribeResponseTest.java
new file mode 100644
index 000000000000..bd15bd30ad71
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5StaleSubscribeResponseTest.java
@@ -0,0 +1,182 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.v5;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.PulsarChannelInitializer;
+import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.broker.service.scalable.ConsumerAssignment;
+import org.apache.pulsar.broker.service.scalable.ScalableTopicService;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.v5.SegmentRouter.ActiveSegment;
+import org.apache.pulsar.common.api.proto.ScalableConsumerType;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.AutoScalePolicyOverride;
+import org.apache.pulsar.common.policies.data.ScalableTopicStats;
+import org.awaitility.Awaitility;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+/**
+ * A member's subscribe response must not override a newer assignment the controller pushed
+ * meanwhile. The coordinator computes a registering member's assignment under its monitor, but
+ * the response used to be written only once the registration completed; a second member that
+ * registered in between rebalanced the group and pushed the first member a newer assignment with
+ * the same layout epoch, which reached the client ahead of the older response.
+ *
+ * <p>Lives in {@code org.apache.pulsar.client.impl.v5} to drive {@link ScalableConsumerClient}
+ * directly and read its assignment. The broker's {@link ServerCnx} holds a member's registration,
+ * as the connection sees it, until the test releases it: that widens the window between the
+ * coordinator computing the assignment and the connection seeing the registration complete.
+ */
+public class V5StaleSubscribeResponseTest extends MockedPulsarServiceBaseTest {
+
+ private record HeldRegistration(CompletableFuture<Void> registered, CompletableFuture<Void> release) {
+ }
+
+ /** Registrations to hold, by consumer name. */
+ private final Map<String, HeldRegistration> heldRegistrations = new ConcurrentHashMap<>();
+
+ @BeforeClass
+ @Override
+ protected void setup() throws Exception {
+ // The scalable consumer session needs a binary-protocol client.
+ isTcpLookup = true;
+ super.internalSetup();
+ setupDefaultTenantAndNamespace();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void releaseHeldRegistrations() {
+ heldRegistrations.values().forEach(held -> held.release().complete(null));
+ heldRegistrations.clear();
+ }
+
+ @Override
+ protected BrokerService customizeNewBrokerService(BrokerService brokerService) {
+ brokerService.setPulsarChannelInitializerFactory(
+ (pulsar, opts) -> new PulsarChannelInitializer(pulsar, opts) {
+ @Override
+ protected ServerCnx newServerCnx(PulsarService pulsar, String listenerName)
+ throws Exception {
+ return new HoldingServerCnx(pulsar, listenerName);
+ }
+ });
+ return brokerService;
+ }
+
+ private class HoldingServerCnx extends ServerCnx {
+
+ HoldingServerCnx(PulsarService pulsar, String listenerName) {
+ super(pulsar, listenerName);
+ }
+
+ @Override
+ protected CompletableFuture<ConsumerAssignment> registerScalableConsumer(
+ ScalableTopicService scalableTopicService, TopicName topicName, String subscription,
+ String consumerName, long consumerId, ScalableConsumerType consumerType) {
+ CompletableFuture<ConsumerAssignment> registration = super.registerScalableConsumer(
+ scalableTopicService, topicName, subscription, consumerName, consumerId, consumerType);
+ HeldRegistration held = heldRegistrations.get(consumerName);
+ if (held == null) {
+ return registration;
+ }
+ // Completes once the coordinator has registered the consumer and computed its
+ // assignment; the connection sees the registration only when the test releases it.
+ registration.thenRun(() -> held.registered().complete(null));
+ return registration.thenCompose(assignment -> held.release().thenApply(__ -> assignment));
+ }
+ }
+
+ @Test(timeOut = 60_000)
+ public void testLateSubscribeResponseDoesNotOverrideRebalance() throws Exception {
+ String topic = "topic://public/default/stale-subscribe-response-"
+ + UUID.randomUUID().toString().substring(0, 8);
+ admin.scalableTopics().createScalableTopic(topic, 2);
+ // Pin the layout: only the two registrations below may change the assignment.
+ admin.scalableTopics().setAutoScalePolicy(topic,
+ AutoScalePolicyOverride.builder().enabled(false).build());
+ String group = "group";
+ HeldRegistration memberAHeld = new HeldRegistration(new CompletableFuture<>(), new CompletableFuture<>());
+ heldRegistrations.put("member-a", memberAHeld);
+
+ PulsarClientImpl v4Client = (PulsarClientImpl) pulsarClient;
+ @Cleanup
+ ScalableConsumerClient memberA = new ScalableConsumerClient(v4Client, TopicName.get(topic), group,
+ "member-a", ScalableConsumerType.CHECKPOINT);
+ CompletableFuture<List<ActiveSegment>> memberAStart = memberA.start();
+ memberAHeld.registered().get(30, TimeUnit.SECONDS);
+
+ // member-b joins while member-a's registration is held: the coordinator rebalances and
+ // pushes member-a its share of the two segments.
+ @Cleanup
+ ScalableConsumerClient memberB = new ScalableConsumerClient(v4Client, TopicName.get(topic), group,
+ "member-b", ScalableConsumerType.CHECKPOINT);
+ memberB.start().get(30, TimeUnit.SECONDS);
+
+ memberAHeld.release().complete(null);
+ memberAStart.get(30, TimeUnit.SECONDS);
+
+ Map<String, Set<Long>> controllerView = assignedSegmentIds(topic, group);
+ assertThat(controllerView.get("member-a")).as("controller assignment of member-a").hasSize(1);
+ assertThat(controllerView.get("member-b")).as("controller assignment of member-b").hasSize(1);
+ Awaitility.await().untilAsserted(() -> {
+ assertThat(segmentIds(memberA.currentAssignment())).as("member-a assignment")
+ .isEqualTo(controllerView.get("member-a"));
+ assertThat(segmentIds(memberB.currentAssignment())).as("member-b assignment")
+ .isEqualTo(controllerView.get("member-b"));
+ });
+ }
+
+ private Map<String, Set<Long>> assignedSegmentIds(String topic, String group) throws Exception {
+ ScalableTopicStats.SubscriptionStats sub = admin.scalableTopics().getStats(topic)
+ .getSubscriptions().get(group);
+ Map<String, Set<Long>> assigned = new HashMap<>();
+ for (ScalableTopicStats.ConsumerStats consumer : sub.getConsumers()) {
+ assigned.put(consumer.getConsumerName(), new HashSet<>(consumer.getSegmentIds()));
+ }
+ return assigned;
+ }
+
+ private static Set<Long> segmentIds(List<ActiveSegment> segments) {
+ return segments.stream().map(ActiveSegment::segmentId).collect(Collectors.toSet());
+ }
+}