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/13 04:41:44 UTC
(pulsar) 01/06: [fix][client] Avoid duplicate reconcile retries and queue layout stalls in V5 consumers (#26874)
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 5b509a5c668e12b88d3625daf9c89238cee2ea55
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Mon Oct 12 14:59:17 2026 -0700
[fix][client] Avoid duplicate reconcile retries and queue layout stalls in V5 consumers (#26874)
(cherry picked from commit 1ed5a81b43874a4d1411f57d208e242d6fabee06)
---
.../client/impl/v5/ScalableQueueConsumer.java | 90 ++++--
.../client/impl/v5/ScalableStreamConsumer.java | 75 ++++-
.../v5/ScalableQueueConsumerReconcileTest.java | 291 ++++++++++++++++++
.../v5/ScalableStreamConsumerReconcileTest.java | 341 +++++++++++++++++++++
4 files changed, 756 insertions(+), 41 deletions(-)
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableQueueConsumer.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableQueueConsumer.java
index 085dfdedec10..02e40c15078d 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableQueueConsumer.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableQueueConsumer.java
@@ -107,8 +107,17 @@ final class ScalableQueueConsumer<T> implements QueueConsumerImpl<T>, DagWatchCl
/** Most recent layout target. Reconciles always converge toward this value. */
private volatile ClientSegmentLayout latestLayout;
- /** Coalesces concurrent reconcile attempts; only one runs at a time. */
+ /** Set when the segment consumers must be converged onto {@link #latestLayout} again. */
+ private final AtomicBoolean reconcileRequested = new AtomicBoolean(false);
+ /** Held while a layout is applied, so only one is applied at a time. */
private final AtomicBoolean reconcileInProgress = new AtomicBoolean(false);
+ /** Whether a retry of the segments that failed to subscribe is scheduled. Guarded by {@code this}. */
+ private boolean retryScheduled = false;
+ /** Whether the backoff was reset after the pending retry was scheduled. Guarded by {@code this}. */
+ private boolean backoffResetSinceRetry = false;
+ /** Identifies the latest retry: a retry it replaced does nothing. Guarded by {@code this}. */
+ private long retryGeneration = 0;
+ /** Paces the retries. Guarded by {@code this}. */
private final Backoff reconcileBackoff = Backoff.builder()
.initialDelay(Duration.ofMillis(100))
.maxBackoff(Duration.ofSeconds(30))
@@ -308,43 +317,68 @@ final class ScalableQueueConsumer<T> implements QueueConsumerImpl<T>, DagWatchCl
@Override
public void onLayoutChange(ClientSegmentLayout newLayout, ClientSegmentLayout oldLayout) {
// Fully async: safe to run on the netty IO thread that delivered the update.
- // Store the target and kick off a reconcile. If a reconcile is already running,
- // it will re-read latestLayout once it finishes. On failure, the reconcile
- // reschedules itself with exponential backoff until it succeeds (or we close).
latestLayout = newLayout;
reconcile();
}
+ /**
+ * Converge the per-segment consumers onto {@link #latestLayout}. A layout is applied as soon as
+ * it arrives, not after the consumers of the previous one have subscribed: a subscribe that keeps
+ * reconnecting must not delay subscribing the segments a split creates. A consumer that fails to
+ * subscribe is evicted and retried with backoff.
+ */
private void reconcile() {
- if (closed) {
- return;
- }
- if (!reconcileInProgress.compareAndSet(false, true)) {
- // A reconcile is already in progress; it will observe the updated latestLayout
- // when it finishes and re-run if needed.
- return;
+ reconcileRequested.set(true);
+ // One thread applies at a time, and applies again if asked to meanwhile.
+ while (!closed && reconcileRequested.get() && reconcileInProgress.compareAndSet(false, true)) {
+ try {
+ reconcileRequested.set(false);
+ evictFailedSegmentConsumers();
+ subscribeSegments(latestLayout).whenComplete((__, ex) -> {
+ if (ex == null) {
+ synchronized (this) {
+ reconcileBackoff.reset();
+ backoffResetSinceRetry = retryScheduled;
+ }
+ } else if (!closed) {
+ scheduleRetry(ex);
+ }
+ });
+ } finally {
+ reconcileInProgress.set(false);
+ }
}
- ClientSegmentLayout target = latestLayout;
- subscribeSegments(target).whenComplete((__, ex) -> {
- reconcileInProgress.set(false);
- if (closed) {
+ }
+
+ /**
+ * Retry the segments that failed to subscribe after a backoff, with at most one retry scheduled.
+ * A retry scheduled before the backoff was last reset is replaced, so a new failure doesn't wait
+ * out the longer delay of earlier ones.
+ */
+ private void scheduleRetry(Throwable ex) {
+ Duration delay;
+ long generation;
+ synchronized (this) {
+ if (retryScheduled && !backoffResetSinceRetry) {
return;
}
- if (ex == null) {
- reconcileBackoff.reset();
- // If a newer layout arrived during this reconcile, run again to converge.
- if (latestLayout != target) {
- reconcile();
+ retryScheduled = true;
+ backoffResetSinceRetry = false;
+ generation = ++retryGeneration;
+ delay = reconcileBackoff.next();
+ }
+ log.warn().attr("delayMs", delay.toMillis()).exceptionMessage(ex)
+ .log("Failed to apply layout update, retrying after backoff");
+ scheduler().schedule(() -> {
+ synchronized (this) {
+ if (generation != retryGeneration) {
+ // Replaced by a later retry.
+ return;
}
- return;
+ retryScheduled = false;
}
- // Evict failed subscribe futures so the next attempt can re-try them.
- evictFailedSegmentConsumers();
- Duration delay = reconcileBackoff.next();
- log.warn().attr("delayMs", delay.toMillis()).exceptionMessage(ex)
- .log("Failed to apply layout update, retrying after backoff");
- scheduler().schedule(this::reconcile, delay.toMillis(), TimeUnit.MILLISECONDS);
- });
+ reconcile();
+ }, delay.toMillis(), TimeUnit.MILLISECONDS);
}
private void evictFailedSegmentConsumers() {
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java
index e48d901c3779..9764b27876c0 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java
@@ -182,6 +182,13 @@ final class ScalableStreamConsumer<T>
private volatile List<ActiveSegment> latestAssignment;
/** Coalesces concurrent reconcile attempts; only one runs at a time. */
private final AtomicBoolean reconcileInProgress = new AtomicBoolean(false);
+ /** Whether a reconcile retry is scheduled. Guarded by {@code this}. */
+ private boolean retryScheduled = false;
+ /** Whether the backoff was reset after the pending retry was scheduled. Guarded by {@code this}. */
+ private boolean backoffResetSinceRetry = false;
+ /** Identifies the latest retry: a retry it replaced does nothing. Guarded by {@code this}. */
+ private long retryGeneration = 0;
+ /** Paces the retries. Guarded by {@code this}. */
private final Backoff reconcileBackoff = Backoff.builder()
.initialDelay(Duration.ofMillis(100))
.maxBackoff(Duration.ofSeconds(30))
@@ -542,7 +549,9 @@ final class ScalableStreamConsumer<T>
CompletableFuture<Void> result) {
subscribeAssigned(assigned).whenComplete((__, ex) -> {
if (ex == null) {
- reconcileBackoff.reset();
+ synchronized (this) {
+ reconcileBackoff.reset();
+ }
result.complete(null);
return;
}
@@ -556,7 +565,10 @@ final class ScalableStreamConsumer<T>
return;
}
evictFailedSegmentConsumers();
- Duration delay = reconcileBackoff.next();
+ Duration delay;
+ synchronized (this) {
+ delay = reconcileBackoff.next();
+ }
log.info().attr("delayMs", delay.toMillis()).exceptionMessage(ex)
.log("Initial subscribe rejected during rebalance, retrying after backoff");
// Retry with the session's current assignment, not the one this attempt started with: the
@@ -578,11 +590,18 @@ final class ScalableStreamConsumer<T>
* consumer may not have released a segment or bucket range yet, so our subscribe is rejected
* ({@code ConsumerBusy} on Exclusive, {@code ConsumerAssignError} on overlapping STICKY
* ranges) — and neither is retried at the v4 layer.
+ *
+ * <p>The next reconcile starts only once this one's releases have drained and its subscribes
+ * have settled. A segment's release state (drain bookkeeping, receive epoch, acked positions)
+ * is keyed by segment id, so applying a newer assignment meanwhile could re-subscribe, or
+ * release again, a segment that is still draining.
*/
private void reconcile() {
if (closed || !reconcileInProgress.compareAndSet(false, true)) {
return;
}
+ // Evict failed subscribe futures so this attempt re-tries them.
+ evictFailedSegmentConsumers();
List<ActiveSegment> target = latestAssignment;
subscribeAssigned(target).whenComplete((__, ex) -> {
reconcileInProgress.set(false);
@@ -590,22 +609,52 @@ final class ScalableStreamConsumer<T>
return;
}
if (ex == null) {
- reconcileBackoff.reset();
- // If a newer assignment arrived during this reconcile, run again to converge.
- if (latestAssignment != target) {
- reconcile();
+ synchronized (this) {
+ reconcileBackoff.reset();
+ backoffResetSinceRetry = retryScheduled;
}
- return;
}
- // Evict failed subscribe futures so the next attempt can re-try them.
- evictFailedSegmentConsumers();
- Duration delay = reconcileBackoff.next();
- log.warn().attr("delayMs", delay.toMillis()).exceptionMessage(ex)
- .log("Failed to apply assignment update, retrying after backoff");
- scheduler().schedule(this::reconcile, delay.toMillis(), TimeUnit.MILLISECONDS);
+ // If a newer assignment arrived during this reconcile, run again to converge: that run also
+ // re-tries what failed in this one.
+ if (latestAssignment != target) {
+ reconcile();
+ } else if (ex != null) {
+ scheduleRetry(ex);
+ }
});
}
+ /**
+ * Retry the reconcile after a backoff, with at most one retry scheduled. A retry scheduled before
+ * the backoff was last reset is replaced, so a new failure doesn't wait out the longer delay of
+ * earlier ones.
+ */
+ private void scheduleRetry(Throwable ex) {
+ Duration delay;
+ long generation;
+ synchronized (this) {
+ if (retryScheduled && !backoffResetSinceRetry) {
+ return;
+ }
+ retryScheduled = true;
+ backoffResetSinceRetry = false;
+ generation = ++retryGeneration;
+ delay = reconcileBackoff.next();
+ }
+ log.warn().attr("delayMs", delay.toMillis()).exceptionMessage(ex)
+ .log("Failed to apply assignment update, retrying after backoff");
+ scheduler().schedule(() -> {
+ synchronized (this) {
+ if (generation != retryGeneration) {
+ // Replaced by a later retry.
+ return;
+ }
+ retryScheduled = false;
+ }
+ reconcile();
+ }, delay.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
private void evictFailedSegmentConsumers() {
for (var entry : segmentConsumers.entrySet()) {
var future = entry.getValue();
diff --git a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/ScalableQueueConsumerReconcileTest.java b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/ScalableQueueConsumerReconcileTest.java
new file mode 100644
index 000000000000..982a947dd1d5
--- /dev/null
+++ b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/ScalableQueueConsumerReconcileTest.java
@@ -0,0 +1,291 @@
+/*
+ * 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.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import com.google.common.util.concurrent.MoreExecutors;
+import io.netty.util.Timer;
+import java.lang.reflect.Proxy;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BiFunction;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.PulsarClientException.AlreadyClosedException;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.client.util.ScheduledExecutorProvider;
+import org.apache.pulsar.common.api.proto.ScalableTopicDAG;
+import org.apache.pulsar.common.api.proto.SegmentState;
+import org.apache.pulsar.common.naming.TopicName;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class ScalableQueueConsumerReconcileTest {
+
+ private static final TopicName TOPIC = TopicName.get("topic://public/default/reconcile-test");
+
+ /** Segment subscribe attempts per segment. */
+ private final Map<Long, Integer> attempts = new HashMap<>();
+ /** Scheduled retries, run by the test. */
+ private final ArrayDeque<Runnable> retries = new ArrayDeque<>();
+ /** The delay of each retry scheduled, in milliseconds. */
+ private final List<Long> retryDelays = new ArrayList<>();
+ private ExecutorService externalExecutor;
+ private PulsarClientImpl v4;
+ private PulsarClientV5 client;
+
+ @BeforeMethod
+ public void setup() {
+ attempts.clear();
+ retries.clear();
+ retryDelays.clear();
+ externalExecutor = MoreExecutors.newDirectExecutorService();
+ v4 = mock(PulsarClientImpl.class);
+ client = mock(PulsarClientV5.class);
+ ExecutorProvider external = mock(ExecutorProvider.class);
+ ScheduledExecutorProvider scheduledProvider = mock(ScheduledExecutorProvider.class);
+ ScheduledExecutorService scheduler = mock(ScheduledExecutorService.class);
+ when(client.v4Client()).thenReturn(v4);
+ when(v4.externalExecutorProvider()).thenReturn(external);
+ when(external.getExecutor()).thenReturn(externalExecutor);
+ when(v4.timer()).thenReturn(mock(Timer.class));
+ when(v4.getScheduledExecutorProvider()).thenReturn(scheduledProvider);
+ when(scheduledProvider.getExecutor()).thenReturn(scheduler);
+ when(scheduler.schedule(any(Runnable.class), anyLong(), any())).thenAnswer(invocation -> {
+ retries.add(invocation.getArgument(0));
+ retryDelays.add(invocation.getArgument(1));
+ return null;
+ });
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void cleanup() {
+ externalExecutor.shutdown();
+ }
+
+ @Test
+ public void keepsAtMostOneRetryScheduled() throws Exception {
+ // Segment 1's subscribe keeps failing with a transient error.
+ subscribeSegments((segmentId, attempt) -> segmentId == 1
+ ? CompletableFuture.failedFuture(new PulsarClientException("transient"))
+ : CompletableFuture.completedFuture(consumer()));
+ DagWatchClient.LayoutChangeListener[] listener = new DagWatchClient.LayoutChangeListener[1];
+ ClientSegmentLayout initial = layout(0);
+ ScalableQueueConsumer<byte[]> consumer = create(initial, listener);
+ try {
+ ClientSegmentLayout split = layout(1);
+ listener[0].onLayoutChange(split, initial);
+ assertEquals(attempts.get(1L), 1);
+ assertEquals(retries.size(), 1, "the failed subscribe must be retried");
+
+ // The layout is pushed again while the retry waits: each push retries segment 1 right
+ // away, which fails again, without scheduling a second retry.
+ listener[0].onLayoutChange(split, split);
+ listener[0].onLayoutChange(split, split);
+ assertEquals(attempts.get(1L), 3);
+ assertEquals(retries.size(), 1, "at most one retry must be scheduled");
+
+ retries.poll().run();
+ assertEquals(attempts.get(1L), 4);
+ assertEquals(retries.size(), 1, "the retry must be rescheduled while the subscribe keeps failing");
+ assertEquals(attempts.get(0L), 1, "subscribed segments are not recreated");
+ assertEquals(attempts.get(2L), 1, "subscribed segments are not recreated");
+ } finally {
+ consumer.close();
+ }
+ }
+
+ @Test
+ public void retriesANewFailureWithoutWaitingOutAnEarlierBackoff() throws Exception {
+ // Segment 1's subscribe fails three times, then succeeds; segment 3's always fails.
+ subscribeSegments((segmentId, attempt) -> segmentId == 3 || (segmentId == 1 && attempt <= 3)
+ ? CompletableFuture.failedFuture(new PulsarClientException("transient"))
+ : CompletableFuture.completedFuture(consumer()));
+ DagWatchClient.LayoutChangeListener[] listener = new DagWatchClient.LayoutChangeListener[1];
+ ClientSegmentLayout initial = layout(0);
+ ScalableQueueConsumer<byte[]> consumer = create(initial, listener);
+ try {
+ // Each failed retry doubles the backoff.
+ ClientSegmentLayout split = layout(1);
+ listener[0].onLayoutChange(split, initial);
+ retries.poll().run();
+ retries.poll().run();
+ assertEquals(attempts.get(1L), 3);
+ assertTrue(retryDelays.get(2) >= 300, "the backoff must have grown: " + retryDelays);
+
+ // The layout is pushed again while that retry waits, and segment 1 subscribes. Then segment 2
+ // splits and segment 3 fails to subscribe: it's retried after the initial delay, not after the
+ // longer one of the retry still pending.
+ listener[0].onLayoutChange(split, split);
+ assertEquals(attempts.get(1L), 4);
+ listener[0].onLayoutChange(layout(2), split);
+ assertEquals(retries.size(), 2);
+ assertTrue(retryDelays.get(3) < 200, "a new failure must be retried soon: " + retryDelays);
+
+ retries.poll().run();
+ assertEquals(attempts.get(3L), 1, "a replaced retry must do nothing");
+ retries.poll().run();
+ assertEquals(attempts.get(3L), 2);
+ assertEquals(retries.size(), 1);
+ } finally {
+ consumer.close();
+ }
+ }
+
+ @Test
+ public void subscribesNewSegmentsWhileAnotherSubscribeIsPending() throws Exception {
+ // Segment 1's subscribe doesn't complete, as while it keeps reconnecting.
+ CompletableFuture<Consumer<byte[]>> segment1Subscribe = new CompletableFuture<>();
+ subscribeSegments((segmentId, attempt) -> segmentId == 1 ? segment1Subscribe
+ : CompletableFuture.completedFuture(consumer()));
+ DagWatchClient.LayoutChangeListener[] listener = new DagWatchClient.LayoutChangeListener[1];
+ ClientSegmentLayout initial = layout(0);
+ ScalableQueueConsumer<byte[]> consumer = create(initial, listener);
+ try {
+ // Segment 0 splits into 1 and 2, then segment 2 splits into 3 and 4 while segment 1's
+ // subscribe is still pending.
+ ClientSegmentLayout split = layout(1);
+ listener[0].onLayoutChange(split, initial);
+ listener[0].onLayoutChange(layout(2), split);
+ assertEquals(attempts.get(3L), 1, "a new segment must be subscribed right away");
+ assertEquals(attempts.get(4L), 1, "a new segment must be subscribed right away");
+ assertEquals(attempts.get(1L), 1, "an in-flight subscribe must be kept, not duplicated");
+ } finally {
+ segment1Subscribe.complete(consumer());
+ consumer.close();
+ }
+ }
+
+ @Test
+ public void closesASegmentThatLeftTheLayoutOnceItsSubscribeCompletes() throws Exception {
+ // Segment 2's subscribe doesn't complete until the test completes it.
+ CompletableFuture<Consumer<byte[]>> segment2Subscribe = new CompletableFuture<>();
+ subscribeSegments((segmentId, attempt) -> segmentId == 2 ? segment2Subscribe
+ : CompletableFuture.completedFuture(consumer()));
+ DagWatchClient.LayoutChangeListener[] listener = new DagWatchClient.LayoutChangeListener[1];
+ ClientSegmentLayout initial = layout(0);
+ ScalableQueueConsumer<byte[]> consumer = create(initial, listener);
+ try {
+ // Segment 0 splits into 1 and 2, and segment 2 leaves the layout while its subscribe is pending.
+ ClientSegmentLayout split = layout(1);
+ listener[0].onLayoutChange(split, initial);
+ listener[0].onLayoutChange(layout(3), split);
+ AtomicBoolean segment2Closed = new AtomicBoolean();
+ segment2Subscribe.complete(consumer(() -> segment2Closed.set(true)));
+ assertTrue(segment2Closed.get(), "a segment that left the layout must be closed once subscribed");
+ assertEquals(attempts.get(2L), 1);
+ } finally {
+ consumer.close();
+ }
+ }
+
+ /** Answer each segment subscribe with {@code subscribe}, given the segment and its attempt number. */
+ private void subscribeSegments(BiFunction<Long, Integer, CompletableFuture<Consumer<byte[]>>> subscribe) {
+ when(v4.<byte[]>subscribeSegmentAsync(any(), any())).thenAnswer(invocation -> {
+ ConsumerConfigurationData<byte[]> conf = invocation.getArgument(0);
+ String topic = conf.getTopicNames().iterator().next();
+ long segmentId = Long.parseLong(topic.substring(topic.lastIndexOf('-') + 1));
+ return subscribe.apply(segmentId, attempts.merge(segmentId, 1, Integer::sum));
+ });
+ }
+
+ private ScalableQueueConsumer<byte[]> create(ClientSegmentLayout initial,
+ DagWatchClient.LayoutChangeListener[] listener) {
+ DagWatchClient watch = mock(DagWatchClient.class);
+ when(watch.topicName()).thenReturn(TOPIC);
+ doAnswer(invocation -> {
+ listener[0] = invocation.getArgument(0);
+ return null;
+ }).when(watch).setListener(any());
+ ConsumerConfigurationData<byte[]> conf = new ConsumerConfigurationData<>();
+ conf.setSubscriptionName("test");
+ return ScalableQueueConsumer.createAsyncImpl(client, Schema.bytes(), conf, watch, initial, null, null)
+ .join();
+ }
+
+ /**
+ * Segment 0; after the first split, sealed segment 0 with children 1 and 2; after the second,
+ * segment 2 sealed too, with children 3 and 4; once the sealed segments are garbage collected,
+ * only 1, 3 and 4.
+ */
+ private static ClientSegmentLayout layout(long epoch) {
+ ScalableTopicDAG dag = new ScalableTopicDAG().setEpoch(epoch);
+ if (epoch < 3) {
+ dag.addSegment().setSegmentId(0).setHashStart(0).setHashEnd(0xffff)
+ .setState(epoch >= 1 ? SegmentState.SEALED : SegmentState.ACTIVE).setCreatedAtEpoch(0);
+ }
+ if (epoch >= 1) {
+ dag.addSegment().setSegmentId(1).setHashStart(0).setHashEnd(0x7fff)
+ .setState(SegmentState.ACTIVE).setCreatedAtEpoch(1).addParentId(0);
+ }
+ if (epoch >= 1 && epoch < 3) {
+ dag.addSegment().setSegmentId(2).setHashStart(0x8000).setHashEnd(0xffff)
+ .setState(epoch >= 2 ? SegmentState.SEALED : SegmentState.ACTIVE).setCreatedAtEpoch(1)
+ .addParentId(0);
+ }
+ if (epoch >= 2) {
+ dag.addSegment().setSegmentId(3).setHashStart(0x8000).setHashEnd(0xbfff)
+ .setState(SegmentState.ACTIVE).setCreatedAtEpoch(2).addParentId(2);
+ dag.addSegment().setSegmentId(4).setHashStart(0xc000).setHashEnd(0xffff)
+ .setState(SegmentState.ACTIVE).setCreatedAtEpoch(2).addParentId(2);
+ }
+ return ClientSegmentLayout.fromProto(dag, TOPIC);
+ }
+
+ /** A segment consumer with nothing to receive: its pending receive never completes. */
+ private static Consumer<byte[]> consumer() {
+ return consumer(() -> {
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Consumer<byte[]> consumer(Runnable onClose) {
+ CompletableFuture<Object> receive = new CompletableFuture<>();
+ return (Consumer<byte[]>) Proxy.newProxyInstance(Consumer.class.getClassLoader(),
+ new Class<?>[]{Consumer.class}, (proxy, method, args) -> switch (method.getName()) {
+ case "receiveAsync" -> receive;
+ case "closeAsync" -> {
+ onClose.run();
+ receive.completeExceptionally(new AlreadyClosedException("closed"));
+ yield CompletableFuture.completedFuture(null);
+ }
+ case "hashCode" -> System.identityHashCode(proxy);
+ case "equals" -> proxy == args[0];
+ case "toString" -> "consumer";
+ default -> throw new UnsupportedOperationException(method.getName());
+ });
+ }
+}
diff --git a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumerReconcileTest.java b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumerReconcileTest.java
new file mode 100644
index 000000000000..a4b28ece7945
--- /dev/null
+++ b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumerReconcileTest.java
@@ -0,0 +1,341 @@
+/*
+ * 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.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import com.google.common.util.concurrent.MoreExecutors;
+import io.netty.util.Timer;
+import java.lang.reflect.Proxy;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.function.BiFunction;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClientException.AlreadyClosedException;
+import org.apache.pulsar.client.api.PulsarClientException.ConsumerBusyException;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.pulsar.client.impl.v5.SegmentRouter.ActiveSegment;
+import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.client.util.ScheduledExecutorProvider;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.scalable.HashRange;
+import org.apache.pulsar.common.scalable.SegmentTopicName;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class ScalableStreamConsumerReconcileTest {
+
+ private static final TopicName TOPIC = TopicName.get("topic://public/default/reconcile-test");
+
+ /** Segment subscribe attempts per segment. */
+ private final Map<Long, Integer> attempts = new HashMap<>();
+ /** Scheduled reconcile retries, run by the test. */
+ private final ArrayDeque<Runnable> retries = new ArrayDeque<>();
+ /** The delay of each reconcile retry scheduled, in milliseconds. */
+ private final List<Long> retryDelays = new ArrayList<>();
+ /** Scheduled release drain polls, run by the test. */
+ private final ArrayDeque<Runnable> drainPolls = new ArrayDeque<>();
+ /** Messages handed to the application. */
+ private final List<MessageV5<byte[]>> delivered = new ArrayList<>();
+ private ExecutorService externalExecutor;
+ private PulsarClientImpl v4;
+ private PulsarClientV5 client;
+ private ScalableConsumerClient session;
+ private ScalableConsumerClient.AssignmentChangeListener listener;
+
+ @BeforeMethod
+ public void setup() {
+ attempts.clear();
+ retries.clear();
+ retryDelays.clear();
+ drainPolls.clear();
+ delivered.clear();
+ externalExecutor = MoreExecutors.newDirectExecutorService();
+ v4 = mock(PulsarClientImpl.class);
+ client = mock(PulsarClientV5.class);
+ ExecutorProvider external = mock(ExecutorProvider.class);
+ ScheduledExecutorProvider scheduledProvider = mock(ScheduledExecutorProvider.class);
+ ScheduledExecutorService scheduler = mock(ScheduledExecutorService.class);
+ when(client.v4Client()).thenReturn(v4);
+ when(v4.externalExecutorProvider()).thenReturn(external);
+ when(external.getExecutor()).thenReturn(externalExecutor);
+ when(v4.timer()).thenReturn(mock(Timer.class));
+ when(v4.getConfiguration()).thenReturn(new ClientConfigurationData());
+ when(v4.getScheduledExecutorProvider()).thenReturn(scheduledProvider);
+ when(scheduledProvider.getExecutor()).thenReturn(scheduler);
+ // The reconcile retries are scheduled as Runnables, the release drain polls as Callables.
+ when(scheduler.schedule(any(Runnable.class), anyLong(), any())).thenAnswer(invocation -> {
+ retries.add(invocation.getArgument(0));
+ retryDelays.add(invocation.getArgument(1));
+ return null;
+ });
+ when(scheduler.schedule(any(Callable.class), anyLong(), any())).thenAnswer(invocation -> {
+ Callable<?> task = invocation.getArgument(0);
+ drainPolls.add(() -> {
+ try {
+ task.call();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ return null;
+ });
+ session = mock(ScalableConsumerClient.class);
+ doAnswer(invocation -> {
+ listener = invocation.getArgument(0);
+ return null;
+ }).when(session).setListener(any());
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void cleanup() {
+ externalExecutor.shutdown();
+ }
+
+ @Test
+ public void keepsAtMostOneRetryScheduled() throws Exception {
+ // Segment 1's subscribe keeps being rejected, as while its previous owner still holds it.
+ subscribeSegments((segmentId, attempt) -> segmentId == 1
+ ? CompletableFuture.failedFuture(new ConsumerBusyException("busy"))
+ : CompletableFuture.completedFuture(consumer(new CompletableFuture<>())));
+ ScalableStreamConsumer<byte[]> consumer = create(List.of(segment(0)));
+ try {
+ listener.onAssignmentChange(List.of(segment(0), segment(1)), List.of(segment(0)));
+ assertEquals(attempts.get(1L), 1);
+ assertEquals(retries.size(), 1, "the rejected subscribe must be retried");
+
+ // The assignment is pushed again while the retry waits: each push retries segment 1
+ // right away, which fails again, without scheduling a second retry.
+ listener.onAssignmentChange(List.of(segment(0), segment(1)), List.of(segment(0), segment(1)));
+ listener.onAssignmentChange(List.of(segment(0), segment(1)), List.of(segment(0), segment(1)));
+ assertEquals(attempts.get(1L), 3);
+ assertEquals(retries.size(), 1, "at most one retry must be scheduled");
+
+ retries.poll().run();
+ assertEquals(attempts.get(1L), 4);
+ assertEquals(retries.size(), 1, "the retry must be rescheduled while the subscribe keeps failing");
+ } finally {
+ consumer.close();
+ }
+ }
+
+ @Test
+ public void retriesANewFailureWithoutWaitingOutAnEarlierBackoff() throws Exception {
+ // Segment 1's subscribe is rejected three times, then accepted; segment 2's is always rejected.
+ subscribeSegments((segmentId, attempt) -> segmentId == 2 || (segmentId == 1 && attempt <= 3)
+ ? CompletableFuture.failedFuture(new ConsumerBusyException("busy"))
+ : CompletableFuture.completedFuture(consumer(new CompletableFuture<>())));
+ ScalableStreamConsumer<byte[]> consumer = create(List.of(segment(0)));
+ try {
+ // Each failed retry doubles the backoff.
+ listener.onAssignmentChange(List.of(segment(0), segment(1)), List.of(segment(0)));
+ retries.poll().run();
+ retries.poll().run();
+ assertEquals(attempts.get(1L), 3);
+ assertTrue(retryDelays.get(2) >= 300, "the backoff must have grown: " + retryDelays);
+
+ // The assignment is pushed again while that retry waits, and segment 1 is accepted. Then
+ // segment 2 is assigned and rejected: it's retried after the initial delay, not after the
+ // longer one of the retry still pending.
+ listener.onAssignmentChange(List.of(segment(0), segment(1)), List.of(segment(0), segment(1)));
+ assertEquals(attempts.get(1L), 4);
+ listener.onAssignmentChange(List.of(segment(0), segment(1), segment(2)),
+ List.of(segment(0), segment(1)));
+ assertEquals(retries.size(), 2);
+ assertTrue(retryDelays.get(3) < 200, "a new failure must be retried soon: " + retryDelays);
+
+ retries.poll().run();
+ assertEquals(attempts.get(2L), 1, "a replaced retry must do nothing");
+ retries.poll().run();
+ assertEquals(attempts.get(2L), 2);
+ assertEquals(retries.size(), 1);
+ } finally {
+ consumer.close();
+ }
+ }
+
+ @Test
+ public void keepsTheRetryThatFiresWhileASegmentDrains() throws Exception {
+ // Segment 1's subscribe keeps being rejected; segment 0's consumer hands the application a message.
+ CompletableFuture<Message<byte[]>> segment0Receive = new CompletableFuture<>();
+ subscribeSegments((segmentId, attempt) -> segmentId == 1
+ ? CompletableFuture.failedFuture(new ConsumerBusyException("busy"))
+ : CompletableFuture.completedFuture(consumer(segmentId == 0 ? segment0Receive
+ : new CompletableFuture<>())));
+ ScalableStreamConsumer<byte[]> consumer = create(List.of(segment(0)));
+ try {
+ segment0Receive.complete(message());
+ listener.onAssignmentChange(List.of(segment(0), segment(1)), List.of(segment(0)));
+ assertEquals(retries.size(), 1);
+
+ // Segment 0 moves to another member while the application holds its message, so the
+ // reconcile releasing it waits for the ack: the retry fires meanwhile, and does nothing.
+ listener.onAssignmentChange(List.of(segment(1)), List.of(segment(0), segment(1)));
+ retries.poll().run();
+ assertEquals(attempts.get(1L), 2);
+ assertEquals(retries.size(), 0);
+
+ // Once the release drained, the reconcile's own failure schedules the next retry.
+ consumer.acknowledgeCumulative(delivered.get(0).id());
+ runDrainPolls();
+ assertEquals(retries.size(), 1, "the rejected subscribe must still be retried");
+ } finally {
+ consumer.close();
+ }
+ }
+
+ @Test
+ public void appliesAnAssignmentThatArrivedDuringAFailedReconcileRightAway() throws Exception {
+ // Segment 1's subscribe keeps being rejected; segment 0's consumer hands the application a message.
+ CompletableFuture<Message<byte[]>> segment0Receive = new CompletableFuture<>();
+ subscribeSegments((segmentId, attempt) -> segmentId == 1
+ ? CompletableFuture.failedFuture(new ConsumerBusyException("busy"))
+ : CompletableFuture.completedFuture(consumer(segmentId == 0 ? segment0Receive
+ : new CompletableFuture<>())));
+ ScalableStreamConsumer<byte[]> consumer = create(List.of(segment(0)));
+ try {
+ segment0Receive.complete(message());
+
+ // Segment 0 moves to another member while the application holds its message, and segment 1,
+ // assigned instead, is rejected: the reconcile completes once the release drained. Segment 2
+ // is assigned meanwhile.
+ listener.onAssignmentChange(List.of(segment(1)), List.of(segment(0)));
+ listener.onAssignmentChange(List.of(segment(1), segment(2)), List.of(segment(1)));
+ consumer.acknowledgeCumulative(delivered.get(0).id());
+ runDrainPolls();
+ assertEquals(attempts.get(2L), 1, "a newer assignment must be applied right after a failed reconcile");
+ assertEquals(retries.size(), 1, "the rejected subscribe must still be retried");
+ } finally {
+ consumer.close();
+ }
+ }
+
+ @Test
+ public void resubscribesASegmentOnlyAfterItsReleaseDrained() throws Exception {
+ // Segment 0's first consumer hands the application a message; its next one, another.
+ CompletableFuture<Message<byte[]>> firstReceive = new CompletableFuture<>();
+ CompletableFuture<Message<byte[]>> secondReceive = new CompletableFuture<>();
+ subscribeSegments((segmentId, attempt) -> CompletableFuture.completedFuture(consumer(
+ segmentId != 0 ? new CompletableFuture<>() : attempt == 1 ? firstReceive : secondReceive)));
+ ScalableStreamConsumer<byte[]> consumer = create(List.of(segment(0)));
+ try {
+ firstReceive.complete(message());
+ assertEquals(delivered.size(), 1);
+
+ // Segment 0 moves to another member and back while the application holds its message:
+ // the assignment that gives it back is applied once the release has drained.
+ listener.onAssignmentChange(List.of(segment(1)), List.of(segment(0)));
+ listener.onAssignmentChange(List.of(segment(0), segment(1)), List.of(segment(1)));
+ assertEquals(attempts.get(0L), 1, "a segment must not be subscribed again while it drains");
+
+ consumer.acknowledgeCumulative(delivered.get(0).id());
+ runDrainPolls();
+ assertEquals(attempts.get(0L), 2, "the segment must be subscribed again once its release drained");
+ secondReceive.complete(message());
+ assertEquals(delivered.size(), 2, "the new consumer of the segment must deliver");
+ } finally {
+ consumer.close();
+ }
+ }
+
+ /** Run the release drain polls, including those they schedule, until none is left. */
+ private void runDrainPolls() {
+ for (int i = 0; i < 100 && !drainPolls.isEmpty(); i++) {
+ drainPolls.poll().run();
+ }
+ assertEquals(drainPolls.size(), 0);
+ }
+
+ /** Answer each segment subscribe with {@code subscribe}, given the segment and its attempt number. */
+ private void subscribeSegments(BiFunction<Long, Integer, CompletableFuture<Consumer<byte[]>>> subscribe) {
+ when(v4.<byte[]>subscribeSegmentAsync(any(), any())).thenAnswer(invocation -> {
+ ConsumerConfigurationData<byte[]> conf = invocation.getArgument(0);
+ String topic = conf.getTopicNames().iterator().next();
+ long segmentId = Long.parseLong(topic.substring(topic.lastIndexOf('-') + 1));
+ return subscribe.apply(segmentId, attempts.merge(segmentId, 1, Integer::sum));
+ });
+ }
+
+ private ScalableStreamConsumer<byte[]> create(List<ActiveSegment> initialAssignment) {
+ ConsumerConfigurationData<byte[]> conf = new ConsumerConfigurationData<>();
+ conf.setSubscriptionName("test");
+ return ScalableStreamConsumer.createAsyncImpl(client, Schema.bytes(), conf, session, TOPIC.toString(),
+ initialAssignment, msg -> {
+ delivered.add(msg);
+ return CompletableFuture.completedFuture(null);
+ }).join();
+ }
+
+ /** A segment this consumer owns whole. */
+ private static ActiveSegment segment(long segmentId) {
+ HashRange range = HashRange.of(0, 0xffff);
+ return new ActiveSegment(segmentId, range,
+ SegmentTopicName.fromParent(TOPIC, range, segmentId).toString(), null, List.of(), List.of());
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Message<byte[]> message() {
+ Message<byte[]> message = mock(Message.class);
+ when(message.getMessageId()).thenReturn(MessageId.earliest);
+ return message;
+ }
+
+ /** A segment consumer whose first receive is {@code firstReceive}; later receives never complete. */
+ @SuppressWarnings("unchecked")
+ private static Consumer<byte[]> consumer(CompletableFuture<Message<byte[]>> firstReceive) {
+ List<CompletableFuture<Message<byte[]>>> receives = new ArrayList<>();
+ return (Consumer<byte[]>) Proxy.newProxyInstance(Consumer.class.getClassLoader(),
+ new Class<?>[]{Consumer.class}, (proxy, method, args) -> switch (method.getName()) {
+ case "receiveAsync" -> {
+ CompletableFuture<Message<byte[]>> receive =
+ receives.isEmpty() ? firstReceive : new CompletableFuture<>();
+ receives.add(receive);
+ yield receive;
+ }
+ case "acknowledgeCumulativeAsync" -> CompletableFuture.completedFuture(null);
+ case "closeAsync" -> {
+ receives.forEach(receive -> receive.completeExceptionally(
+ new AlreadyClosedException("closed")));
+ yield CompletableFuture.completedFuture(null);
+ }
+ case "hashCode" -> System.identityHashCode(proxy);
+ case "equals" -> proxy == args[0];
+ case "toString" -> "consumer";
+ default -> throw new UnsupportedOperationException(method.getName());
+ });
+ }
+}