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/19 07:09:40 UTC
(pulsar) branch master updated: [fix][client] Keep per-key order of V5 async sends across scalable topic layout changes (#26820)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new cb82bd3175a6 [fix][client] Keep per-key order of V5 async sends across scalable topic layout changes (#26820)
cb82bd3175a6 is described below
commit cb82bd3175a67794041c4fe84a92f4e57d06cde9
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Oct 19 10:09:31 2026 +0300
[fix][client] Keep per-key order of V5 async sends across scalable topic layout changes (#26820)
---
.../client/api/v5/V5ProducerAccessModeTest.java | 32 +
.../v5/V5ProducerLayoutChangeOrderingTest.java | 202 ++++++
.../client/impl/v5/V5ProducerSegmentGoneTest.java | 35 +
.../pulsar/client/impl/v5/ProducerBuilderV5.java | 1 +
.../pulsar/client/impl/v5/PulsarClientV5.java | 39 +-
.../client/impl/v5/ScalableTopicProducer.java | 759 ++++++++++++++++++---
.../impl/v5/ScalableTopicProducerOrderingTest.java | 571 ++++++++++++++++
7 files changed, 1561 insertions(+), 78 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ProducerAccessModeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ProducerAccessModeTest.java
index b4b466ab97fa..8067be9ded88 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ProducerAccessModeTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ProducerAccessModeTest.java
@@ -18,9 +18,14 @@
*/
package org.apache.pulsar.client.api.v5;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.expectThrows;
import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
import lombok.Cleanup;
import org.apache.pulsar.client.api.v5.config.ProducerAccessMode;
import org.apache.pulsar.client.api.v5.schema.Schema;
@@ -122,4 +127,31 @@ public class V5ProducerAccessModeTest extends V5ClientBaseTest {
MessageId id = second.newMessage().value("after-takeover").send();
assertNotNull(id);
}
+
+ @Test
+ public void testClosingClientWithWaitingExclusiveProducerCompletes() throws Exception {
+ String topic = newScalableTopic(1);
+
+ @Cleanup
+ Producer<String> incumbent = v5Client.newProducer(Schema.string())
+ .topic(topic)
+ .accessMode(ProducerAccessMode.EXCLUSIVE)
+ .create();
+
+ // Another client waits for exclusive access, which it never gets: closing that client must
+ // not wait for it.
+ PulsarClient waitingClient = newV5Client();
+ CompletableFuture<Producer<String>> waiting = waitingClient.newProducer(Schema.string())
+ .topic(topic)
+ .accessMode(ProducerAccessMode.WAIT_FOR_EXCLUSIVE)
+ .createAsync();
+ Thread.sleep(500);
+ assertFalse(waiting.isDone(), "WAIT_FOR_EXCLUSIVE create() must wait while another producer holds the claim");
+
+ waitingClient.closeAsync().get(10, TimeUnit.SECONDS);
+ assertThatThrownBy(() -> waiting.get(10, TimeUnit.SECONDS)).isInstanceOf(ExecutionException.class);
+
+ // The incumbent keeps its claim.
+ assertNotNull(incumbent.newMessage().value("still-exclusive").send());
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerLayoutChangeOrderingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerLayoutChangeOrderingTest.java
new file mode 100644
index 000000000000..342dfd4ac113
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerLayoutChangeOrderingTest.java
@@ -0,0 +1,202 @@
+/*
+ * 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.time.Duration;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.LongStream;
+import lombok.Cleanup;
+import org.apache.pulsar.client.api.v5.Message;
+import org.apache.pulsar.client.api.v5.MessageId;
+import org.apache.pulsar.client.api.v5.Producer;
+import org.apache.pulsar.client.api.v5.StreamConsumer;
+import org.apache.pulsar.client.api.v5.V5ClientBaseTest;
+import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.common.policies.data.ScalableTopicMetadata;
+import org.testng.annotations.Test;
+
+/**
+ * Per-key order of async sends that are in flight while the topic layout changes.
+ *
+ * <p>When a split or merge seals a segment, the sends still pending on it fail with
+ * {@code TopicTerminated} / {@code AlreadyClosed} and are re-sent to the successor segments.
+ * Meanwhile the producer keeps routing new sends to those successors. Unless the re-sends go
+ * first, a newer message for a key is stored before an older one, and every consumer sees that key
+ * out of order. These tests keep an async producer busy across the layout change and then check the
+ * order the messages of each key were stored in: segment by segment along the key's lineage in the
+ * segment DAG, and in each segment in the order it was read.
+ */
+public class V5ProducerLayoutChangeOrderingTest extends V5ClientBaseTest {
+
+ private static final int KEYS = 200;
+ private static final int MESSAGES_BEFORE_CHANGE = 2_000;
+ private static final int MESSAGES_AFTER_CHANGE = 2_000;
+ /** Sends per burst; a burst is followed by a millisecond pause, a pace the broker keeps up with. */
+ private static final int BURST = 20;
+
+ @Test
+ public void testAsyncSendsKeepPerKeyOrderAcrossSplit() throws Exception {
+ String topic = newScalableTopic(1);
+ produceAndVerifyPerKeyOrder(topic, () -> {
+ List<Long> active = activeSegmentIds(topic);
+ admin.scalableTopics().splitSegment(topic, active.get(0));
+ });
+ }
+
+ @Test
+ public void testAsyncSendsKeepPerKeyOrderAcrossMerge() throws Exception {
+ String topic = newScalableTopic(2);
+ produceAndVerifyPerKeyOrder(topic, () -> {
+ List<Long> active = activeSegmentIds(topic);
+ admin.scalableTopics().mergeSegments(topic, active.get(0), active.get(1));
+ });
+ }
+
+ @Test
+ public void testAsyncSendsKeepPerKeyOrderAcrossConsecutiveSplits() throws Exception {
+ String topic = newScalableTopic(1);
+ produceAndVerifyPerKeyOrder(topic, () -> {
+ admin.scalableTopics().splitSegment(topic, activeSegmentIds(topic).get(0));
+ // Split a child right away, while the sends re-sent after the first split may still be
+ // in flight to it.
+ admin.scalableTopics().splitSegment(topic, activeSegmentIds(topic).get(0));
+ });
+ }
+
+ private interface LayoutChange {
+ void run() throws Exception;
+ }
+
+ /**
+ * Sends {@code key -> 1, 2, 3, ...} for {@link #KEYS} keys with async sends, runs the layout
+ * change while the sends keep going, and checks that every key was stored as exactly
+ * {@code 1..n}.
+ */
+ private void produceAndVerifyPerKeyOrder(String topic, LayoutChange change) throws Exception {
+ // Subscribe first, so that every message is retained for the subscription.
+ @Cleanup
+ StreamConsumer<Long> consumer = v5Client.newStreamConsumer(Schema.int64())
+ .topic(topic)
+ .subscriptionName("ordering-sub")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe();
+ @Cleanup
+ Producer<Long> producer = v5Client.newProducer(Schema.int64())
+ .topic(topic)
+ .create();
+
+ long[] nextSequence = new long[KEYS];
+ List<CompletableFuture<MessageId>> sends = new ArrayList<>();
+ int sent = 0;
+ for (; sent < MESSAGES_BEFORE_CHANGE; sent++) {
+ sends.add(sendNext(producer, nextSequence, sent));
+ }
+
+ // Change the layout while sends keep flowing, so that some are in flight to the sealed
+ // segment and the rest are routed to its successors once the new layout arrives.
+ CompletableFuture<Void> layoutChange = CompletableFuture.runAsync(() -> {
+ try {
+ change.run();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ while (!layoutChange.isDone()) {
+ sends.add(sendNext(producer, nextSequence, sent++));
+ if (sent % BURST == 0) {
+ Thread.sleep(1);
+ }
+ }
+ layoutChange.get(30, TimeUnit.SECONDS);
+ for (int i = 0; i < MESSAGES_AFTER_CHANGE; i++) {
+ sends.add(sendNext(producer, nextSequence, sent++));
+ if (sent % BURST == 0) {
+ Thread.sleep(1);
+ }
+ }
+ CompletableFuture.allOf(sends.toArray(CompletableFuture[]::new)).get(60, TimeUnit.SECONDS);
+ // Otherwise the layout change caught no send in flight, and the test proved nothing.
+ assertThat(((ScalableTopicProducer<Long>) producer).sendsParkedOrHeld.get())
+ .as("sends parked or held behind the retiring segment").isPositive();
+
+ // Read until the topic is drained, so that a duplicate shows up as such instead of hiding the
+ // last message of another key. Per key, keep the sequence numbers of each segment in the
+ // order they were read, which is the order that segment stores them in.
+ Map<String, Map<Long, List<Long>>> stored = new HashMap<>();
+ int count = 0;
+ Message<Long> msg;
+ while ((msg = consumer.receive(Duration.ofSeconds(count < sent ? 10 : 2))) != null) {
+ long segmentId = ((MessageIdV5) msg.id()).segmentId();
+ stored.computeIfAbsent(msg.key().orElseThrow(), __ -> new HashMap<>())
+ .computeIfAbsent(segmentId, __ -> new ArrayList<>())
+ .add(msg.value());
+ consumer.acknowledgeCumulative(msg.id());
+ count++;
+ }
+ assertThat(count).as("messages received").isEqualTo(sent);
+
+ // A key's segments are a path in the DAG, each created after the one it came from.
+ Map<Long, ScalableTopicMetadata.SegmentInfo> segments =
+ admin.scalableTopics().getMetadata(topic).getSegments();
+ for (int k = 0; k < KEYS; k++) {
+ Map<Long, List<Long>> bySegment = stored.get(key(k));
+ assertThat(bySegment).as("messages of %s", key(k)).isNotNull();
+ List<Long> inStoredOrder = new ArrayList<>();
+ Map<Long, List<Long>> lineage = new LinkedHashMap<>();
+ bySegment.keySet().stream()
+ .sorted(Comparator.comparingLong((Long id) -> segments.get(id).getCreatedAtEpoch())
+ .thenComparingLong(id -> id))
+ .forEach(id -> {
+ inStoredOrder.addAll(bySegment.get(id));
+ lineage.put(id, bySegment.get(id));
+ });
+ List<Long> expected = LongStream.rangeClosed(1, nextSequence[k]).boxed().toList();
+ assertThat(inStoredOrder)
+ .as("stored order of %s across the layout change, by segment: %s", key(k), lineage)
+ .containsExactlyElementsOf(expected);
+ }
+ }
+
+ private static CompletableFuture<MessageId> sendNext(Producer<Long> producer, long[] nextSequence,
+ int index) {
+ int k = index % KEYS;
+ return producer.async().newMessage().key(key(k)).value(++nextSequence[k]).send();
+ }
+
+ private static String key(int k) {
+ return "key-" + k;
+ }
+
+ private List<Long> activeSegmentIds(String topic) throws Exception {
+ return admin.scalableTopics().getMetadata(topic).getSegments().values().stream()
+ .filter(ScalableTopicMetadata.SegmentInfo::isActive)
+ .map(ScalableTopicMetadata.SegmentInfo::getSegmentId)
+ .sorted()
+ .toList();
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerSegmentGoneTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerSegmentGoneTest.java
index 9c30598cf12a..b0484daaf574 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerSegmentGoneTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerSegmentGoneTest.java
@@ -174,6 +174,41 @@ public class V5ProducerSegmentGoneTest extends V5ClientBaseTest {
assertEquals(producer.segmentProducersCreated.get(), 1);
}
+ @Test
+ public void closingTheClientFailsTheSendsWaitingForTheLayout() throws Exception {
+ PulsarClient client = newClient();
+ String topic = newScalableTopic(1);
+ long segmentId = singleActiveSegmentId(topic);
+ ScalableTopicProducer<byte[]> producer = newProducer(client, topic);
+
+ Sends sends = new Sends();
+ Thread sender = startSender(producer, sends);
+ try {
+ Awaitility.await().until(() -> sends.segmentsAcked.contains(segmentId));
+ // Terminate the segment topic underneath the layout, while quiet (see above). The sends
+ // that fail on it wait for a layout that never comes, and the ones after them are held
+ // behind them: none of them is in flight on a v4 producer that closing the client would
+ // fail.
+ sends.pause.set(true);
+ Awaitility.await().until(sends::drained);
+ admin.scalableTopics().terminateSegment(segmentTopic(topic, segmentId));
+ sends.pause.set(false);
+ Awaitility.await().until(() -> producer.waitingSendCount() > 0);
+ sends.pause.set(true);
+ Awaitility.await().until(() -> producer.inFlightSendCount() == 0);
+ assertEquals(sends.failed.get(), 0, "the sends must still be waiting for the layout");
+ client.close();
+ } finally {
+ stopSender(sender, sends);
+ }
+
+ // Closing the client fails them right away, well before their retry budget would run out.
+ allSettled(sends).get(2, TimeUnit.SECONDS);
+ assertTrue(sends.failed.get() > 0, "expected the waiting sends to fail");
+ assertEquals(((PulsarClientV5) client).v4Client().getMemoryLimitController().currentUsage(), 0,
+ "the failed sends must give their memory back");
+ }
+
@Test
public void closingTheClientWhileSendsAreInFlightFailsThemWithoutRecreatingProducers() throws Exception {
PulsarClient client = newClient();
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ProducerBuilderV5.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ProducerBuilderV5.java
index c244fe9d970a..2c92f368157c 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ProducerBuilderV5.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ProducerBuilderV5.java
@@ -79,6 +79,7 @@ final class ProducerBuilderV5<T> implements ProducerBuilder<T> {
.thenCompose(initialLayout -> {
ScalableTopicProducer<T> producer = new ScalableTopicProducer<>(
client, v5Schema, conf, dagWatch, initialLayout);
+ client.registerProducer(producer);
// For exclusive access modes, claim every active segment up front so
// a collision surfaces here instead of being deferred to first send.
return producer.eagerAttachInitialAsync()
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientV5.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientV5.java
index 599d1fddbd2b..602b2ca9eccc 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientV5.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientV5.java
@@ -19,7 +19,9 @@
package org.apache.pulsar.client.impl.v5;
import java.time.Duration;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.v5.CheckpointConsumerBuilder;
@@ -42,6 +44,14 @@ final class PulsarClientV5 implements PulsarClient {
private final String description;
private final Duration transactionTimeout;
+ /**
+ * The V5 producers on scalable topics, until their close completes. Closing the v4 client closes
+ * the per-segment v4 producers but not these, whose sends waiting for a new layout would then
+ * never complete: they are closed first, and the v4 client once they are, including those already
+ * closing.
+ */
+ private final Set<ScalableTopicProducer<?>> producers = ConcurrentHashMap.newKeySet();
+
PulsarClientV5(PulsarClientImpl v4Client, String description, Duration transactionTimeout) {
this.v4Client = v4Client;
this.description = description;
@@ -55,6 +65,21 @@ final class PulsarClientV5 implements PulsarClient {
return v4Client;
}
+ void registerProducer(ScalableTopicProducer<?> producer) {
+ producers.add(producer);
+ }
+
+ void deregisterProducer(ScalableTopicProducer<?> producer) {
+ producers.remove(producer);
+ }
+
+ /** Close the V5 producers, before the v4 client and its executors. Never fails. */
+ private CompletableFuture<Void> closeProducersAsync() {
+ return CompletableFuture.allOf(producers.stream()
+ .map(producer -> producer.closeAsync().exceptionally(__ -> null))
+ .toArray(CompletableFuture[]::new));
+ }
+
@Override
public <T> ProducerBuilder<T> newProducer(Schema<T> schema) {
return new ProducerBuilderV5<>(this, schema);
@@ -99,6 +124,14 @@ final class PulsarClientV5 implements PulsarClient {
@Override
public void close() throws PulsarClientException {
+ try {
+ closeProducersAsync().get();
+ } catch (InterruptedException e) {
+ // Close the v4 client all the same.
+ Thread.currentThread().interrupt();
+ } catch (ExecutionException e) {
+ // Not reached: closeProducersAsync never fails.
+ }
try {
v4Client.close();
} catch (org.apache.pulsar.client.api.PulsarClientException e) {
@@ -108,11 +141,15 @@ final class PulsarClientV5 implements PulsarClient {
@Override
public CompletableFuture<Void> closeAsync() {
- return v4Client.closeAsync().exceptionallyCompose(V5ExceptionMappings::failWithV5Exception);
+ return closeProducersAsync().thenCompose(__ -> v4Client.closeAsync())
+ .exceptionallyCompose(V5ExceptionMappings::failWithV5Exception);
}
@Override
public void shutdown() {
+ // Without waiting: this fails the sends waiting for a layout right away, and the v4 shutdown
+ // fails the ones in flight.
+ closeProducersAsync();
try {
v4Client.shutdown();
} catch (org.apache.pulsar.client.api.PulsarClientException e) {
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducer.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducer.java
index 9d5010d8b259..0df0340f54c4 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducer.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducer.java
@@ -24,7 +24,10 @@ import io.netty.channel.EventLoopGroup;
import io.netty.util.concurrent.EventExecutor;
import java.time.Duration;
import java.time.Instant;
+import java.util.ArrayDeque;
import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -37,6 +40,8 @@ import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import org.apache.pulsar.client.api.EncodeData;
@@ -73,6 +78,13 @@ import org.apache.pulsar.common.scalable.ScalableTopicConstants;
* producers only account for the payload once a message reaches them and never block on the limit
* themselves ({@link ProducerConfigurationData#isMemoryLimitAdmittedUpstream()}), so nothing on the
* dispatch chain, whose links may run on an IO thread, ever waits for memory.
+ *
+ * <p>Per-key order across layout changes: async sends of a key are stored in call order across a
+ * split or merge (see {@link #dispatchSend}). While a segment retires, new async sends whose key it
+ * owned are held until the sends that were pending on it have been re-sent, which takes until the new
+ * layout reaches the client; sends of other keys, and sends without a key, are not held. A synchronous
+ * {@link MessageBuilder#send()} does not go through the hold: mixing synchronous and async sends of
+ * one key can reorder them across a layout change.
*/
final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.LayoutChangeListener {
@@ -86,6 +98,19 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
/** Failure of a send routed on a layout that changed before its segment producer was created. */
private static final String SEGMENT_NOT_ACTIVE = "not found in active segments";
+ // Per-key ordering states of an async send (PendingSend.state).
+ /** Not tracked: a synchronous send, or an async send that is done or being re-routed. */
+ private static final int SEND_UNTRACKED = 0;
+ /** Routed to a segment and not resolved yet. */
+ private static final int SEND_IN_FLIGHT = 1;
+ /** Its segment is gone: waits to be re-sent, in call order, once the segment has retired. */
+ private static final int SEND_PARKED = 2;
+ /** A new send held back while a segment retires. */
+ private static final int SEND_HELD = 3;
+
+ /** {@link #dispatchSend} argument for a send from the caller, rather than released. */
+ private static final long NOT_RELEASED = -1;
+
/**
* Memory charged for each pending message on top of its payload, covering the per-message
* bookkeeping the client retains until the send completes: message metadata, the v4 send op and
@@ -120,9 +145,10 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
/**
* Where the caller's future is completed when its result did not arrive on an IO thread: on
- * the dispatch thread, the timer, the thread closing the producer. Code chained on the future
- * may block, and a send from it may wait for memory, which none of those threads can afford to
- * do (see {@link #finish}). Results that arrive on an IO thread complete in place.
+ * the dispatch thread or the timer. Code chained on the future may block, and a send from it may
+ * wait for memory, which neither thread can afford to do (see {@link #finish}). Results that
+ * arrive on an IO thread complete in place, and so do all results once closing, when a send
+ * from the chained code fails fast.
*/
private final ExecutorService completionExecutor;
@@ -137,6 +163,10 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
@VisibleForTesting
final AtomicInteger segmentProducersCreated = new AtomicInteger();
+ /** Test hook: how many times an async send was parked or held behind a retiring segment. */
+ @VisibleForTesting
+ final AtomicLong sendsParkedOrHeld = new AtomicLong();
+
/**
* Per-segment v4 producers. Stored as futures so concurrent send-on-cold-segment
* calls share a single creation attempt without blocking, and so callers running
@@ -162,6 +192,20 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
dispatchChains = new ConcurrentHashMap<>();
private final Object dispatchLock = new Object();
+ /**
+ * Tails of the dispatch chains of the segments a layout dropped, until they complete.
+ * {@link #closeAsync} waits for them along with the current chains.
+ */
+ private final Set<CompletableFuture<?>> droppedChainTails = ConcurrentHashMap.newKeySet();
+
+ /**
+ * The heads of the dispatch chains, until they complete. {@link #closeAsync} fails the ones still
+ * pending, and with them the sends behind them: a head waits for the segment producer to be
+ * created, which under {@code WAIT_FOR_EXCLUSIVE} can take forever, and then for a hop to the
+ * dispatch executor, which the client drops if it shuts down first.
+ */
+ private final Set<CompletableFuture<?>> pendingChainHeads = ConcurrentHashMap.newKeySet();
+
/**
* Currently in-flight async sends. {@link #flushAsync()} snapshots and
* awaits these (each user-visible send future completes on broker ack —
@@ -177,10 +221,67 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
private final Set<CompletableFuture<MessageId>> inFlightSends =
ConcurrentHashMap.newKeySet();
+ /**
+ * Guards the per-key ordering state of async sends: the segment each one is routed to, the
+ * segments being retired, and the sends held back while they are (see {@link #dispatchSend}).
+ * Held only for this bookkeeping and never while calling into a segment producer, so that a v4
+ * callback taking it cannot deadlock against a dispatch holding a v4 lock. Parking a send also
+ * charges the client memory limit under it, which may run the limit's trigger inline; that
+ * trigger does not call back into this producer.
+ */
+ private final Object orderLock = new Object();
+
+ /**
+ * The async sends routed to each segment and not yet resolved. The entry of an active segment
+ * stays while the segment does, so that it is not allocated and dropped as its in-flight count
+ * goes 0 and back; that of a segment the layout dropped goes once it is idle or retired. Guarded
+ * by {@link #orderLock}.
+ */
+ private final Map<Long, SegmentSends<T>> segmentSends = new HashMap<>();
+
+ /** New async sends held, in call order, while a segment retires. Guarded by {@link #orderLock}. */
+ private final ArrayDeque<PendingSend<T>> heldSends = new ArrayDeque<>();
+
+ /**
+ * Whether new async sends are held: set when a segment starts retiring, cleared once the held
+ * sends have all been dispatched. Only the sends of a key that one of {@link #holdScopes} owns
+ * are held. Guarded by {@link #orderLock}.
+ */
+ private boolean holdingSends;
+
+ /**
+ * The segments that started retiring since {@link #holdingSends} was last set, each with the
+ * layout it was active in: a new send is held when its key routes to one of them in that layout,
+ * which are the keys whose earlier sends may still be re-sent. Kept until the hold clears, so
+ * that a key's new sends keep queueing behind its held ones. Guarded by {@link #orderLock}.
+ */
+ private final List<HoldScope> holdScopes = new ArrayList<>();
+
+ /** How many sends were parked and held since the hold was set, for logging. Guarded by {@link #orderLock}. */
+ private int parkedInHold;
+ private int heldInHold;
+
+ /**
+ * How many layouts were applied. {@link #releaseSends} stops dispatching a batch when it changes:
+ * the sends left in the batch would be routed by the new layout, possibly to the successors of a
+ * segment that earlier sends of the batch went to, and overtake them there. Guarded by
+ * {@link #orderLock}.
+ */
+ private long layoutChanges;
+
+ /** Call order of the async sends, by which the sends of a retired segment are re-sent. */
+ private final AtomicLong sendOrder = new AtomicLong();
+
+ /** Serializes {@link #releaseSends}, so that released sends are dispatched in order. */
+ private final Object releaseLock = new Object();
+ private final AtomicBoolean releaseScheduled = new AtomicBoolean();
+
// Current active segments (volatile for visibility across threads)
private volatile List<ActiveSegment> activeSegments = List.of();
private volatile boolean closed = false;
+ /** Completes once {@link #closeAsync} is done; set by its first call. */
+ private final AtomicReference<CompletableFuture<Void>> closeFuture = new AtomicReference<>();
private final AsyncProducerV5<T> asyncView;
ScalableTopicProducer(PulsarClientV5 client,
@@ -291,6 +392,18 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
*/
volatile boolean dispatching;
+ // Ordering state of an async send, guarded by orderLock.
+ /** Position in call order. */
+ long order;
+ /** One of the {@code SEND_*} states. */
+ int state = SEND_UNTRACKED;
+ /** The segment it is in flight to or parked on, in those states. */
+ long segmentId = -1;
+ /** How many times it was parked because its segment was gone. */
+ int attempts;
+ /** Why it was last parked: the failure it ends with if it can't be re-sent. */
+ Throwable lastFailure;
+
PendingSend(String key, V value, EncodeData encoded, int payloadSize,
Map<String, String> properties, Instant eventTime, Long sequenceId,
Duration deliverAfter, Instant deliverAt, List<String> replicationClusters,
@@ -310,6 +423,29 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
}
}
+ /**
+ * The async sends routed to one segment, guarded by {@link #orderLock}. A segment is retiring
+ * once one of its sends failed because the segment is gone, or the layout dropped it while it
+ * had sends: from then on no new send is routed anywhere until its in-flight sends are resolved
+ * and the new layout is in, and its parked sends have been re-sent.
+ */
+ private static final class SegmentSends<V> {
+ int inFlight;
+ final List<PendingSend<V>> parked = new ArrayList<>();
+ boolean retiring;
+
+ boolean isIdle() {
+ return inFlight == 0 && parked.isEmpty() && !retiring;
+ }
+ }
+
+ /**
+ * A retiring segment, with the layout it was active in, or null if that is not known: see
+ * {@link #holdScopes}.
+ */
+ private record HoldScope(long segmentId, List<ActiveSegment> layout) {
+ }
+
/**
* Encode the value on the caller's thread, so that the memory limit can be charged the exact
* payload size before the message is queued, and the schema's CPU work stays off the dispatch
@@ -552,45 +688,69 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
// Tracked until the caller-visible future is done, whichever side completes it. Registered
// before the future is handed out, so it runs after the caller's own continuations.
userFuture.whenComplete((__, ___) -> inFlightSends.remove(userFuture));
- dispatchSendAttempt(send, 0);
+ send.order = sendOrder.getAndIncrement();
+ dispatchSend(send, NOT_RELEASED);
return userFuture;
}
- private void dispatchSendAttempt(PendingSend<T> send, int attempt) {
+ /**
+ * Route an async send and append it to its segment's dispatch chain, keeping per-key order
+ * across layout changes.
+ *
+ * <p>When a segment is sealed by a split/merge, or terminated by a regular-to-scalable
+ * migration, the sends still pending on it fail and must be re-sent to its successors. Newer
+ * sends for the same keys must not reach the successors before them. So once a segment is gone
+ * it retires: its failed sends are parked rather than retried one by one, and new sends of the
+ * keys it owned are held, in call order, instead of routed. When the new layout is in and the
+ * segment has no send in flight any more, {@link #releaseSends} re-sends the parked sends in call
+ * order and then the held ones. Routing and the retirement state change together under
+ * {@link #orderLock}, so a send of such a key either goes to the old segment and is parked if it
+ * fails there, or is held. Sends of other keys, and sends without a key, which have no order to
+ * keep, are routed as usual meanwhile.
+ *
+ * @param releasedAt {@link #NOT_RELEASED} for a send from the caller, which is held while new
+ * sends are held; for a send {@link #releaseSends} dispatches, the
+ * {@link #layoutChanges} count its batch was taken at: such a send is routed
+ * even while new sends are held, unless a new layout was applied since
+ * @return false if a released send was not dispatched because a new layout was applied
+ */
+ private boolean dispatchSend(PendingSend<T> send, long releasedAt) {
if (abandonIfDone(send)) {
- return;
- }
- if (isShuttingDown()) {
- finish(send, null, new PulsarClientException.AlreadyClosedException("Producer already closed"));
- return;
+ return true;
}
long segmentId;
- try {
- segmentId = routeMessage(send.key);
- } catch (Exception e) {
- finish(send, null, e);
- return;
+ Exception failure = null;
+ synchronized (orderLock) {
+ // Checked under the lock: closeAsync fails the held sends under it after closing.
+ if (isShuttingDown()) {
+ failure = new PulsarClientException.AlreadyClosedException("Producer already closed");
+ segmentId = -1;
+ } else if (releasedAt == NOT_RELEASED && holdingSends && isHeldKey(send.key)) {
+ send.state = SEND_HELD;
+ heldSends.add(send);
+ heldInHold++;
+ sendsParkedOrHeld.incrementAndGet();
+ return true;
+ } else if (releasedAt != NOT_RELEASED && releasedAt != layoutChanges) {
+ return false;
+ } else {
+ try {
+ segmentId = routeMessage(send.key);
+ segmentSends.computeIfAbsent(segmentId, __ -> new SegmentSends<>()).inFlight++;
+ send.state = SEND_IN_FLIGHT;
+ send.segmentId = segmentId;
+ } catch (Exception e) {
+ failure = e;
+ segmentId = -1;
+ }
+ }
+ }
+ if (failure != null) {
+ finish(send, null, failure);
+ return true;
}
final long routedSegmentId = segmentId;
- // Re-dispatch this message on the next attempt. Used when the target segment is gone
- // — sealed by a split/merge or terminated by a regular-to-scalable migration — and
- // the DAG watch is expected to refresh the layout shortly so routeMessage lands on an
- // active child. The dead per-segment producer stays where it is until applyLayout drops
- // it: a send routed to the segment in the meantime fails fast on it, instead of
- // re-creating a producer on a terminated topic for every send that finds it gone.
- Runnable retry = () -> {
- if (abandonIfDone(send)) {
- return;
- }
- // The message stays with this layer while it waits for the new layout.
- reholdPayloadShare(send);
- CompletableFuture.delayedExecutor(
- Math.min(100L * (attempt + 1), SEND_RETRY_MAX_BACKOFF_MS),
- TimeUnit.MILLISECONDS)
- .execute(() -> dispatchSendAttempt(send, attempt + 1));
- };
-
appendToDispatchChain(routedSegmentId,
producer -> {
if (abandonIfDone(send)) {
@@ -603,7 +763,7 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
ackFuture = buildV4Message(producer, send).sendAsync();
} catch (Exception e) {
// Only this send fails; the chain stays healthy for the sends behind it.
- finish(send, null, e);
+ onSendFailure(send, routedSegmentId, e);
return;
}
if (afterV4EnqueueHook != null) {
@@ -619,7 +779,7 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
finish(send, new MessageIdV5(v4MsgId, routedSegmentId), null);
} else {
// Failure from the v4 send (e.g. the segment sealed mid-flight).
- handleAsyncSegmentFailure(send, routedSegmentId, attempt, ex, retry);
+ onSendFailure(send, routedSegmentId, ex);
}
});
} finally {
@@ -628,25 +788,375 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
},
// Failure while (re)creating the per-segment producer — e.g. the partition was
// terminated by a migration between routing and creation.
- createEx -> handleAsyncSegmentFailure(send, routedSegmentId, attempt, createEx, retry));
+ createEx -> onSendFailure(send, routedSegmentId, createEx));
+ return true;
}
/**
- * Decide whether an async send failure should be retried. If the target segment is gone
- * (a split/merge seal or a migration termination) and the retry budget isn't exhausted,
- * run {@code retry}; otherwise fail the user-visible future. No new layout is coming once
- * this producer or its client is closing, so their sends fail right away. Covers both the
- * v4 send failure and the per-segment producer-creation failure.
+ * An async send failed on {@code segmentId}. If the segment is gone (a split/merge seal or a
+ * migration termination) and the send has attempts left, park it and retire the segment, to be
+ * re-sent in call order by {@link #releaseSends}; otherwise fail the user-visible future. No new
+ * layout is coming once this producer or its client is closing, so their sends fail right away.
+ * Covers both the v4 send failure and the per-segment producer-creation failure.
*/
- private void handleAsyncSegmentFailure(PendingSend<T> send, long segmentId,
- int attempt, Throwable ex, Runnable retry) {
+ private void onSendFailure(PendingSend<T> send, long segmentId, Throwable ex) {
Throwable cause = ex instanceof CompletionException ? ex.getCause() : ex;
- if (isSegmentGoneError(cause) && attempt < SEND_RETRY_MAX_ATTEMPTS && !isShuttingDown()) {
- log.debug().attr("segmentId", segmentId).attr("attempt", attempt + 1)
- .log("Target segment gone, retrying async send after layout update");
- retry.run();
+ if (!isSegmentGoneError(cause)) {
+ finish(send, null, ex);
+ return;
+ }
+ boolean parked = false;
+ boolean startedRetiring = false;
+ boolean shuttingDown;
+ synchronized (orderLock) {
+ if (send.state != SEND_IN_FLIGHT || send.segmentId != segmentId) {
+ // Already given up on: finish() untracked it.
+ return;
+ }
+ // Checked under the lock, together with parking: failWaitingSends drains under it once
+ // closing is visible, and a send parked after that would wait for a layout that never
+ // comes. No new layout is coming once this producer or its client is closing.
+ shuttingDown = isShuttingDown();
+ if (!shuttingDown) {
+ SegmentSends<T> sends = segmentSends.get(segmentId);
+ sends.inFlight--;
+ if (send.attempts < SEND_RETRY_MAX_ATTEMPTS) {
+ send.attempts++;
+ send.lastFailure = ex;
+ send.state = SEND_PARKED;
+ sends.parked.add(send);
+ // The message stays with this layer until it is re-sent.
+ reholdPayloadShare(send);
+ parkedInHold++;
+ sendsParkedOrHeld.incrementAndGet();
+ parked = true;
+ } else {
+ send.state = SEND_UNTRACKED;
+ send.segmentId = -1;
+ }
+ if (!sends.retiring) {
+ // The layout still has the segment: a layout that dropped it while this send
+ // was in flight would have started its retirement.
+ startRetiring(segmentId, sends, activeSegments);
+ startedRetiring = true;
+ }
+ }
+ }
+ if (shuttingDown) {
+ finish(send, null, ex);
+ // Closing the client does not close this producer: fail what waits for a layout here.
+ failWaitingSends();
+ return;
+ }
+ if (parked) {
+ log.debug().attr("segmentId", segmentId).attr("attempt", send.attempts)
+ .log("Target segment gone, re-sending async send in order after layout update");
} else {
- finish(send, null, V5ExceptionMappings.toV5Exception(ex));
+ finish(send, null, ex);
+ }
+ if (startedRetiring) {
+ // The segment failed before the layout dropped it: wait for the new layout.
+ scheduleRetirementCheck(segmentId, 1);
+ }
+ scheduleRelease();
+ }
+
+ private void scheduleRelease() {
+ if (!releaseScheduled.compareAndSet(false, true)) {
+ return;
+ }
+ try {
+ dispatchExecutor.execute(this::releaseSends);
+ } catch (RejectedExecutionException e) {
+ // The client is closing and its executors are gone: nothing will release the waiting sends.
+ releaseScheduled.set(false);
+ failWaitingSends();
+ }
+ }
+
+ /**
+ * No new layout is coming once this producer or its client is closing: fail the sends waiting
+ * for one, parked or held. Closing the client does not close this producer, so besides
+ * {@link #closeAsync}, every path that finds the client closing calls this. Idempotent. The sends
+ * still in flight fail as their segment producers close.
+ */
+ private void failWaitingSends() {
+ List<PendingSend<T>> waiting = new ArrayList<>();
+ synchronized (orderLock) {
+ waiting.addAll(heldSends);
+ heldSends.clear();
+ for (var sends : segmentSends.values()) {
+ waiting.addAll(sends.parked);
+ sends.parked.clear();
+ }
+ for (var send : waiting) {
+ send.state = SEND_UNTRACKED;
+ send.segmentId = -1;
+ }
+ clearHold();
+ }
+ for (var send : waiting) {
+ finish(send, null, new PulsarClientException.AlreadyClosedException("Producer already closed"));
+ }
+ }
+
+ /**
+ * Re-send the parked sends of the segments that have retired, in call order, and once no
+ * segment is retiring any more, the held sends. A segment has retired when the layout no longer
+ * has it and none of its sends is in flight. Runs on the dispatch executor, one at a time, and
+ * dispatches outside {@link #orderLock}: a send failing right away re-enters it.
+ *
+ * <p>A batch stops when a new layout is applied while it is dispatched: the rest of the batch
+ * would be routed by it, possibly to the successors of a segment its first sends went to. The
+ * rest goes back to the head of the held sends: it is older than all of them, and newer than what
+ * was dispatched. A send of the batch failing on a segment that is gone while the layout stays
+ * does not stop it: the rest goes to the same segments, where any that fail are parked too and
+ * re-sent in call order.
+ */
+ private void releaseSends() {
+ synchronized (releaseLock) {
+ releaseScheduled.set(false);
+ if (isShuttingDown()) {
+ failWaitingSends();
+ return;
+ }
+ while (true) {
+ List<PendingSend<T>> batch = new ArrayList<>();
+ long batchLayout;
+ boolean parkedBatch;
+ synchronized (orderLock) {
+ if (!holdingSends) {
+ return;
+ }
+ batchLayout = layoutChanges;
+ boolean retiring = false;
+ var it = segmentSends.entrySet().iterator();
+ while (it.hasNext()) {
+ var entry = it.next();
+ SegmentSends<T> sends = entry.getValue();
+ if (!sends.retiring) {
+ continue;
+ }
+ if (sends.inFlight == 0 && !isActive(entry.getKey())) {
+ batch.addAll(sends.parked);
+ it.remove();
+ } else {
+ retiring = true;
+ }
+ }
+ parkedBatch = !batch.isEmpty();
+ if (!parkedBatch) {
+ if (retiring) {
+ return;
+ }
+ if (heldSends.isEmpty()) {
+ clearHold();
+ return;
+ }
+ batch.addAll(heldSends);
+ heldSends.clear();
+ }
+ for (var send : batch) {
+ send.state = SEND_UNTRACKED;
+ send.segmentId = -1;
+ }
+ }
+ if (parkedBatch) {
+ // Retired segments of a merge interleave: restore the call order. The order of a
+ // send is set before it is dispatched and never changes.
+ batch.sort(Comparator.comparingLong(send -> send.order));
+ }
+ for (int i = 0; i < batch.size(); i++) {
+ if (!dispatchSend(batch.get(i), batchLayout)) {
+ boolean closing;
+ synchronized (orderLock) {
+ // failWaitingSends fails the held sends under the lock after closing.
+ closing = isShuttingDown();
+ if (!closing) {
+ for (int j = batch.size() - 1; j >= i; j--) {
+ var send = batch.get(j);
+ send.state = SEND_HELD;
+ heldSends.addFirst(send);
+ }
+ }
+ }
+ if (closing) {
+ for (int j = i; j < batch.size(); j++) {
+ finish(batch.get(j), null,
+ new PulsarClientException.AlreadyClosedException("Producer already closed"));
+ }
+ failWaitingSends();
+ return;
+ }
+ break;
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * A segment whose sends failed because it is gone, while the layout still has it, retires once
+ * the new layout drops it. If no new layout comes within the retry budget, its parked sends fail.
+ *
+ * <p>Runs on the JDK's delayed executor rather than the client's: the client's executors drop
+ * queued tasks when they shut down, as on {@link PulsarClientV5#shutdown()}, and the check must
+ * still run then to fail the waiting sends. It only does bookkeeping and completes futures, which
+ * {@link #finish} hands off.
+ */
+ private void scheduleRetirementCheck(long segmentId, int check) {
+ CompletableFuture.delayedExecutor(Math.min(100L * check, SEND_RETRY_MAX_BACKOFF_MS),
+ TimeUnit.MILLISECONDS)
+ .execute(() -> checkRetirement(segmentId, check));
+ }
+
+ private void checkRetirement(long segmentId, int check) {
+ if (isShuttingDown()) {
+ failWaitingSends();
+ return;
+ }
+ List<PendingSend<T>> expired = List.of();
+ boolean recheck = false;
+ synchronized (orderLock) {
+ SegmentSends<T> sends = segmentSends.get(segmentId);
+ if (sends == null || !sends.retiring || !isActive(segmentId)) {
+ // Retired: releaseSends re-sends its parked sends.
+ } else if (check < SEND_RETRY_MAX_ATTEMPTS || sends.inFlight > 0) {
+ recheck = true;
+ } else {
+ expired = new ArrayList<>(sends.parked);
+ for (var send : expired) {
+ send.state = SEND_UNTRACKED;
+ send.segmentId = -1;
+ }
+ segmentSends.remove(segmentId);
+ }
+ }
+ if (recheck) {
+ scheduleRetirementCheck(segmentId, check + 1);
+ return;
+ }
+ for (var send : expired) {
+ finish(send, null, send.lastFailure);
+ }
+ scheduleRelease();
+ }
+
+ /** Test hook: how many async sends wait to be re-sent or dispatched, parked or held. */
+ @VisibleForTesting
+ int waitingSendCount() {
+ synchronized (orderLock) {
+ int waiting = heldSends.size();
+ for (var sends : segmentSends.values()) {
+ waiting += sends.parked.size();
+ }
+ return waiting;
+ }
+ }
+
+ /** Test hook: how many async sends are in flight to a segment. */
+ @VisibleForTesting
+ int inFlightSendCount() {
+ synchronized (orderLock) {
+ int inFlight = 0;
+ for (var sends : segmentSends.values()) {
+ inFlight += sends.inFlight;
+ }
+ return inFlight;
+ }
+ }
+
+ /**
+ * Start retiring a segment, whose keys are those it owns in {@code layout}: hold the new sends
+ * of those keys. Called under {@link #orderLock}.
+ */
+ private void startRetiring(long segmentId, SegmentSends<T> sends, List<ActiveSegment> layout) {
+ sends.retiring = true;
+ boolean inLayout = layout.stream().anyMatch(segment -> segment.segmentId() == segmentId);
+ // Without the layout it was active in, its keys are unknown: hold the sends of every key.
+ holdScopes.add(new HoldScope(segmentId, inLayout ? layout : null));
+ if (!holdingSends) {
+ holdingSends = true;
+ log.debug().attr("segmentId", segmentId)
+ .log("Segment retiring: holding the async sends of its keys");
+ }
+ }
+
+ /** Stop holding sends, once none is held or parked. Called under {@link #orderLock}. */
+ private void clearHold() {
+ if (holdingSends) {
+ log.debug().attr("parked", parkedInHold).attr("held", heldInHold)
+ .log("Retired segments done: no longer holding async sends");
+ }
+ holdingSends = false;
+ holdScopes.clear();
+ parkedInHold = 0;
+ heldInHold = 0;
+ }
+
+ /**
+ * Whether a new send of {@code key} is held while sends are: whether one of {@link #holdScopes}
+ * owns the key. Sends without a key have no order to keep. Called under {@link #orderLock}.
+ */
+ private boolean isHeldKey(String key) {
+ if (key == null) {
+ return false;
+ }
+ for (var scope : holdScopes) {
+ if (scope.layout() == null) {
+ return true;
+ }
+ try {
+ if (router.route(key, scope.layout()) == scope.segmentId()) {
+ return true;
+ }
+ } catch (IllegalStateException e) {
+ // Not routable in that layout, which does not happen for a layout that was applied:
+ // hold the send rather than risk its order.
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private boolean isActive(long segmentId) {
+ for (var segment : activeSegments) {
+ if (segment.segmentId() == segmentId) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /** Drop a finished async send from the ordering state. */
+ private void untrack(PendingSend<T> send) {
+ boolean release = false;
+ synchronized (orderLock) {
+ switch (send.state) {
+ case SEND_IN_FLIGHT -> {
+ SegmentSends<T> sends = segmentSends.get(send.segmentId);
+ sends.inFlight--;
+ release = sends.retiring;
+ if (sends.isIdle() && !isActive(send.segmentId)) {
+ segmentSends.remove(send.segmentId);
+ }
+ }
+ // The two cases below are not reached today: every path that finishes a parked or
+ // held send untracks it first. Guards for a future path that does not.
+ case SEND_PARKED -> {
+ SegmentSends<T> sends = segmentSends.get(send.segmentId);
+ sends.parked.remove(send);
+ release = true;
+ }
+ case SEND_HELD -> heldSends.remove(send);
+ default -> {
+ }
+ }
+ send.state = SEND_UNTRACKED;
+ send.segmentId = -1;
+ }
+ if (release) {
+ scheduleRelease();
}
}
@@ -661,23 +1171,31 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
* queued behind it, nor on the timer or the thread closing the producer. A result delivered
* inside the dispatch link, whether a failure raised in {@code sendAsync} or an acknowledgement
* that beat the callback registration, is on the dispatch thread. Those complete on
- * {@link #completionExecutor}.
+ * {@link #completionExecutor}, unless this producer or the client is closing: then they complete
+ * in place, as nothing may be left queued on an executor the client is about to shut down.
+ *
+ * <p>A failure is delivered as the V5 exception of its error category, whichever path failed
+ * the send.
*/
- private void finish(PendingSend<T> send, MessageId messageId, Throwable failure) {
+ private void finish(PendingSend<T> send, MessageId messageId, Throwable sendFailure) {
if (!send.finished.compareAndSet(false, true)) {
return;
}
+ Throwable failure = sendFailure != null ? V5ExceptionMappings.toV5Exception(sendFailure) : null;
+ untrack(send);
releaseAll(send);
if (send.userFuture.isDone()) {
// Completed by the caller (cancelled, timed out): nothing left to deliver.
return;
}
boolean offload = send.dispatching || (failure != null && !isEventLoopThread());
- if (!offload || client.v4Client().isClosed()) {
- // In place: the result arrived on an IO thread, or the client is closing. Its executors
- // are shut down with it, with shutdownNow(), which drops whatever is still queued: a
- // completion handed to the executor now could be lost and the caller's future never
- // complete. Once the executors are gone, the same happens below.
+ if (!offload || isShuttingDown()) {
+ // In place: the result arrived on an IO thread, or this producer or the client is
+ // closing. The client's executors are shut down with it, with shutdownNow(), which drops
+ // whatever is still queued: a completion handed to the executor now could be lost and the
+ // caller's future never complete. Closing the client closes this producer first, so a
+ // completion queued while this producer closes could be lost the same way. Once the
+ // executors are gone, the same happens below.
complete(send, messageId, failure);
return;
}
@@ -722,21 +1240,42 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
Consumer<org.apache.pulsar.client.api.Producer<T>> dispatchOp,
Consumer<Throwable> onCreateFailure) {
synchronized (dispatchLock) {
- // The head completes on the IO thread that finishes creating the segment producer, and
- // every link queued behind it would run there in one burst: hop off it first.
- var prev = dispatchChains.computeIfAbsent(segmentId,
- id -> getOrCreateSegmentProducerAsync(id)
- .thenApplyAsync(Function.identity(), dispatchExecutor));
- var next = prev.thenApply(producer -> {
- dispatchOp.accept(producer);
- return producer;
- });
- // If the chain link itself faults (creation failure), hand it to the caller.
- next.exceptionally(ex -> {
- onCreateFailure.accept(ex);
- return null;
- });
- dispatchChains.put(segmentId, next);
+ if (!closed) {
+ // The head completes on the IO thread that finishes creating the segment producer, and
+ // every link queued behind it would run there in one burst: hop off it first.
+ var prev = dispatchChains.computeIfAbsent(segmentId, id -> {
+ var head = getOrCreateSegmentProducerAsync(id)
+ .thenApplyAsync(Function.identity(), dispatchExecutor);
+ pendingChainHeads.add(head);
+ head.whenComplete((__, ___) -> pendingChainHeads.remove(head));
+ return head;
+ });
+ var next = prev.thenApply(producer -> {
+ dispatchOp.accept(producer);
+ return producer;
+ });
+ // If the chain link itself faults (creation failure), hand it to the caller.
+ next.exceptionally(ex -> {
+ onCreateFailure.accept(ex);
+ return null;
+ });
+ dispatchChains.put(segmentId, next);
+ return;
+ }
+ }
+ // closeAsync may have collected the chain tails already: a link appended now would not be
+ // waited for. Fail the send as gone, which fails it for good once closing.
+ onCreateFailure.accept(new PulsarClientException.AlreadyClosedException("Producer already closed"));
+ }
+
+ /** Drop the dispatch chain of a segment the layout dropped, keeping its tail until it completes. */
+ private void dropDispatchChain(long segmentId) {
+ synchronized (dispatchLock) {
+ var tail = dispatchChains.remove(segmentId);
+ if (tail != null && !tail.isDone()) {
+ droppedChainTails.add(tail);
+ tail.whenComplete((__, ___) -> droppedChainTails.remove(tail));
+ }
}
}
@@ -787,22 +1326,64 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
return CompletableFuture.allOf(pending);
}
+ /**
+ * Close the producer: fail the sends waiting for a layout and those behind a dispatch chain head
+ * that is still pending, close the segment producers, and wait for the sends on the dispatch
+ * chains to be handed to them, which fails them as the segment producers are closed.
+ *
+ * <p>Nothing this waits for depends on the client's executors or on a segment producer still
+ * being created: the client closes its producers first and then shuts its executors down, which
+ * drops their queued tasks, and a segment producer waiting for exclusive access may never be
+ * created. A segment producer still being created is closed once it is, without waiting for it;
+ * closing the client closes it right away. The producer stays registered with the client until
+ * this completes, so that closing the client waits for it too. Idempotent.
+ */
CompletableFuture<Void> closeAsync() {
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ if (!closeFuture.compareAndSet(null, result)) {
+ return closeFuture.get();
+ }
closed = true;
dagWatch.close();
- List<CompletableFuture<Void>> futures = new ArrayList<>();
+ // No new layout is coming: fail the sends waiting for one. The sends in flight fail as their
+ // segment producers close below.
+ failWaitingSends();
+
+ List<CompletableFuture<?>> futures = new ArrayList<>();
for (var future : segmentProducers.values()) {
// If creation failed, there's nothing to close — swallow so a single bad
// segment doesn't fail the overall close.
- futures.add(future.thenCompose(p -> p.closeAsync())
- .exceptionally(__ -> null));
+ var segmentClosed = future.thenCompose(p -> p.closeAsync()).exceptionally(__ -> null);
+ if (future.isDone()) {
+ futures.add(segmentClosed);
+ }
}
- return CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))
+ // Collected after closing is visible, under the lock: no chain is started or link appended
+ // after this.
+ List<CompletableFuture<?>> heads;
+ synchronized (dispatchLock) {
+ heads = new ArrayList<>(pendingChainHeads);
+ for (var tail : dispatchChains.values()) {
+ futures.add(tail.handle((__, ___) -> null));
+ }
+ for (var tail : droppedChainTails) {
+ futures.add(tail.handle((__, ___) -> null));
+ }
+ }
+ // Outside the lock: the sends behind the heads fail in place, and the caller's code chained
+ // on them runs here.
+ for (var head : heads) {
+ head.completeExceptionally(new PulsarClientException.AlreadyClosedException("Producer already closed"));
+ }
+ CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))
.whenComplete((__, ___) -> {
segmentProducers.clear();
dispatchChains.clear();
+ client.deregisterProducer(this);
+ result.complete(null);
});
+ return result;
}
// --- Layout change handling ---
@@ -836,14 +1417,38 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
}
private void applyLayout(ClientSegmentLayout layout) {
- this.activeSegments = layout.activeSegments();
-
// Determine which segments are new and which are gone
Set<Long> newSegmentIds = ConcurrentHashMap.newKeySet();
for (var seg : layout.activeSegments()) {
newSegmentIds.add(seg.segmentId());
}
+ // A segment the layout drops while sends are pending on it retires: new sends are held
+ // until those are resolved and the failed ones re-sent (see dispatchSend). Done together
+ // with the switch to the new layout, so no send is routed by it before the hold is on.
+ boolean holding;
+ synchronized (orderLock) {
+ var it = segmentSends.entrySet().iterator();
+ while (it.hasNext()) {
+ var entry = it.next();
+ SegmentSends<T> sends = entry.getValue();
+ if (newSegmentIds.contains(entry.getKey()) || sends.retiring) {
+ continue;
+ }
+ if (sends.isIdle()) {
+ it.remove();
+ } else {
+ startRetiring(entry.getKey(), sends, activeSegments);
+ }
+ }
+ this.activeSegments = layout.activeSegments();
+ layoutChanges++;
+ holding = holdingSends;
+ }
+ if (holding) {
+ scheduleRelease();
+ }
+
// Close producers for segments that are no longer active
for (var entry : segmentProducers.entrySet()) {
if (!newSegmentIds.contains(entry.getKey())) {
@@ -861,7 +1466,7 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
}
});
segmentProducers.remove(entry.getKey());
- dispatchChains.remove(entry.getKey());
+ dropDispatchChain(entry.getKey());
}
}
diff --git a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducerOrderingTest.java b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducerOrderingTest.java
new file mode 100644
index 000000000000..29b91c1ebeef
--- /dev/null
+++ b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducerOrderingTest.java
@@ -0,0 +1,571 @@
+/*
+ * 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 static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.RETURNS_SELF;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import com.google.common.util.concurrent.MoreExecutors;
+import io.netty.channel.EventLoopGroup;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.AbstractExecutorService;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Predicate;
+import org.apache.pulsar.client.api.EncodeData;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClientException.AlreadyClosedException;
+import org.apache.pulsar.client.api.PulsarClientException.TopicTerminatedException;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.client.impl.MemoryLimitController;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.TypedMessageBuilderImpl;
+import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
+import org.apache.pulsar.client.util.ExecutorProvider;
+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;
+
+/**
+ * The per-key ordering of {@link ScalableTopicProducer}'s async sends across layout changes, driven
+ * step by step: the layouts are applied by the test, the segment producers are fakes whose sends
+ * the test acknowledges or fails, and the dispatch executor runs its tasks on the test thread.
+ *
+ * <p>The layouts: segment 0 covers every key; it splits into segment 1 (the low half of the hash
+ * space) and segment 2 (the high half); segment 1 then splits into segments 3 and 4.
+ */
+public class ScalableTopicProducerOrderingTest {
+
+ private static final TopicName TOPIC = TopicName.get("topic://public/default/ordering-test");
+
+ private static final ClientSegmentLayout L0 = layout(0);
+ private static final ClientSegmentLayout L1 = layout(1);
+ private static final ClientSegmentLayout L2 = layout(2);
+
+ /** A key of segment 3, and so of segments 1 and 0. */
+ private static final String KEY_A = keyIn(0x0000, 0x3fff);
+ /** A key of segment 2, and of segment 0. */
+ private static final String KEY_B = keyIn(0x8000, 0xffff);
+
+ /** A message handed to a segment producer, with the future its result is delivered on. */
+ private record Sent(long segmentId, String key, String value, CompletableFuture<MessageId> result) {
+ }
+
+ private final List<Sent> sent = Collections.synchronizedList(new ArrayList<>());
+ private final Set<Long> closedSegments = ConcurrentHashMap.newKeySet();
+ private final Map<String, Long> segmentIds = new HashMap<>();
+ private TestExecutor dispatchExecutor;
+ private ExecutorService completionExecutor;
+ private ExecutorProvider external;
+ /** What the segment producers' {@code closeAsync} returns. */
+ private volatile CompletableFuture<Void> segmentCloseResult;
+ private MemoryLimitController memory;
+ private PulsarClientImpl v4;
+ private PulsarClientV5 client;
+ private DagWatchClient.LayoutChangeListener listener;
+ private ScalableTopicProducer<String> producer;
+
+ @BeforeMethod
+ public void setup() {
+ sent.clear();
+ closedSegments.clear();
+ segmentCloseResult = CompletableFuture.completedFuture(null);
+ for (var layout : List.of(L0, L1, L2)) {
+ for (var segment : layout.activeSegments()) {
+ segmentIds.put(segment.attachTopicName(), segment.segmentId());
+ }
+ }
+ dispatchExecutor = new TestExecutor();
+ completionExecutor = MoreExecutors.newDirectExecutorService();
+ memory = new MemoryLimitController(64 * 1024 * 1024);
+ v4 = mock(PulsarClientImpl.class);
+ external = mock(ExecutorProvider.class);
+ EventLoopGroup eventLoopGroup = mock(EventLoopGroup.class);
+ when(v4.getMemoryLimitController()).thenReturn(memory);
+ when(v4.getInternalExecutorService()).thenReturn(dispatchExecutor);
+ when(v4.eventLoopGroup()).thenReturn(eventLoopGroup);
+ when(eventLoopGroup.iterator()).thenAnswer(__ -> Collections.emptyIterator());
+ when(v4.externalExecutorProvider()).thenReturn(external);
+ when(external.getExecutor()).thenReturn(completionExecutor);
+ when(v4.createSegmentProducerAsync(any(), any())).thenAnswer(invocation -> {
+ ProducerConfigurationData conf = invocation.getArgument(0);
+ return CompletableFuture.completedFuture(segmentProducer(segmentIds.get(conf.getTopicName())));
+ });
+ when(v4.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));
+ client = new PulsarClientV5(v4, "test", null);
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void cleanup() {
+ if (producer != null) {
+ producer.closeAsync();
+ dispatchExecutor.paused = false;
+ dispatchExecutor.runAll();
+ }
+ completionExecutor.shutdown();
+ }
+
+ @Test
+ public void resendsTheSendsOfASealedSegmentBeforeTheNewerOnes() {
+ create(L0);
+ var a1 = send(KEY_A, "a1");
+ var b1 = send(KEY_B, "b1");
+ assertThat(values(0)).containsExactly("a1", "b1");
+
+ // The split: segment 0 retires, and the new sends of its keys, every key, are held.
+ applyLayout(L1, L0);
+ var a2 = send(KEY_A, "a2");
+ var b2 = send(KEY_B, "b2");
+ assertThat(producer.waitingSendCount()).isEqualTo(2);
+ assertThat(sent).hasSize(2);
+
+ // Segment 0 is sealed: its sends fail, and are re-sent before the held ones.
+ complete(0, new TopicTerminatedException("sealed"));
+ assertThat(values(1)).containsExactly("a1", "a2");
+ assertThat(values(2)).containsExactly("b1", "b2");
+ assertThat(producer.sendsParkedOrHeld.get()).isEqualTo(4);
+
+ complete(1, null);
+ complete(2, null);
+ assertThat(segmentOf(a1)).isEqualTo(1);
+ assertThat(segmentOf(a2)).isEqualTo(1);
+ assertThat(segmentOf(b1)).isEqualTo(2);
+ assertThat(segmentOf(b2)).isEqualTo(2);
+ assertThat(producer.waitingSendCount()).isZero();
+ assertThat(memory.currentUsage()).isZero();
+ }
+
+ @Test
+ public void holdsOnlyTheSendsOfTheKeysOfTheRetiringSegment() {
+ create(L1);
+ var a1 = send(KEY_A, "a1");
+ var b1 = send(KEY_B, "b1");
+
+ // Segment 1 is terminated underneath the layout: a1 waits for the next layout, and so does
+ // a2 behind it.
+ complete(1, new TopicTerminatedException("terminated"));
+ var a2 = send(KEY_A, "a2");
+ assertThat(producer.waitingSendCount()).isEqualTo(2);
+
+ // The sends of segment 2's keys, and those without a key, go out meanwhile.
+ var b2 = send(KEY_B, "b2");
+ assertThat(values(2)).containsExactly("b1", "b2");
+ var keyless = send(null, "k");
+ assertThat(sent).anyMatch(s -> s.value().equals("k"));
+ assertThat(producer.waitingSendCount()).isEqualTo(2);
+ complete(2, null);
+ assertThat(b1).isCompleted();
+ assertThat(b2).isCompleted();
+ // The keyless send, if it went to segment 1, waits for the layout too.
+ complete(1, new TopicTerminatedException("terminated"));
+
+ // Segment 1 splits: its keys' sends go to its children, in order.
+ applyLayout(L2, L1);
+ assertThat(valuesOfKey(3, KEY_A)).containsExactly("a1", "a2");
+ completeAll();
+ assertThat(segmentOf(a1)).isEqualTo(3);
+ assertThat(segmentOf(a2)).isEqualTo(3);
+ assertThat(keyless).isCompleted();
+ assertThat(memory.currentUsage()).isZero();
+ }
+
+ @Test
+ public void aLayoutAppliedWhileReleasingStopsTheRestOfTheBatch() {
+ create(L0);
+ var a1 = send(KEY_A, "a1");
+ var a2 = send(KEY_A, "a2");
+ applyLayout(L1, L0);
+ // Sends without a key are not held: they start the dispatch chains of segments 1 and 2, so
+ // that the re-sent a1 is handed to segment 1's producer while the batch is released.
+ send(null, "k1");
+ send(null, "k2");
+ complete(1, null);
+ complete(2, null);
+ var a3 = send(KEY_A, "a3");
+
+ // Segment 1 splits right after a1 is re-sent to it, before a2 is: a2 must wait for a1 to be
+ // re-sent again, rather than go to segment 3 ahead of it.
+ AtomicBoolean split = new AtomicBoolean();
+ producer.afterV4EnqueueHook = __ -> {
+ Sent last = sent.get(sent.size() - 1);
+ if (last.segmentId() == 1 && last.value().equals("a1") && split.compareAndSet(false, true)) {
+ listener.onLayoutChange(L2, L1);
+ }
+ };
+ complete(0, new TopicTerminatedException("sealed"));
+ assertThat(split).isTrue();
+ assertThat(valuesOfKey(1, KEY_A)).containsExactly("a1");
+ assertThat(values(3)).isEmpty();
+
+ complete(1, new TopicTerminatedException("sealed"));
+ assertThat(values(3)).containsExactly("a1", "a2", "a3");
+ completeAll();
+ assertThat(segmentOf(a1)).isEqualTo(3);
+ assertThat(segmentOf(a2)).isEqualTo(3);
+ assertThat(segmentOf(a3)).isEqualTo(3);
+ assertThat(memory.currentUsage()).isZero();
+ }
+
+ @Test
+ public void closingTheClientFailsTheWaitingSendsWhenTheirReleaseIsDropped() throws Exception {
+ create(L0);
+ client.registerProducer(producer);
+ // Shutting the v4 client down drops the dispatch executor's queued tasks.
+ doAnswer(__ -> {
+ dispatchExecutor.shutdownNow();
+ return null;
+ }).when(v4).close();
+
+ var a1 = send(KEY_A, "a1");
+ applyLayout(L1, L0);
+ var a2 = send(KEY_A, "a2");
+ // a1 fails on the sealed segment while the dispatch executor is behind: the release of a1
+ // and a2 is queued when the client closes.
+ dispatchExecutor.paused = true;
+ complete(0, new TopicTerminatedException("sealed"));
+ assertThat(dispatchExecutor.tasks).isNotEmpty();
+
+ client.close();
+ Class<?> alreadyClosed = org.apache.pulsar.client.api.v5.PulsarClientException.AlreadyClosedException.class;
+ assertThat(a1).failsWithin(0, TimeUnit.SECONDS).withThrowableThat()
+ .havingCause().isInstanceOf(alreadyClosed);
+ assertThat(a2).failsWithin(0, TimeUnit.SECONDS).withThrowableThat()
+ .havingCause().isInstanceOf(alreadyClosed);
+ assertThat(memory.currentUsage()).isZero();
+ }
+
+ @Test
+ public void closingTheClientFailsTheSendsBehindAQueuedChainHop() {
+ create(L0);
+ client.registerProducer(producer);
+ // The send's dispatch chain starts with a hop to the dispatch executor, which is behind.
+ dispatchExecutor.paused = true;
+ var a1 = send(KEY_A, "a1");
+ assertThat(sent).isEmpty();
+
+ // Closing the v4 client would drop the hop: closing the producer fails the send without it.
+ var closed = client.closeAsync();
+ assertThat(closed).isCompleted();
+ verify(v4).closeAsync();
+ assertClosedFailure(a1);
+ assertThat(memory.currentUsage()).isZero();
+
+ dispatchExecutor.paused = false;
+ dispatchExecutor.runAll();
+ assertThat(sent).isEmpty();
+ }
+
+ @Test
+ public void closingTheClientCompletesTheWaitingSendsBeforeItsExecutorsShutDown() {
+ // The completion executor is behind, and closing the v4 client drops its queued tasks.
+ var completions = new TestExecutor();
+ completions.paused = true;
+ completionExecutor = completions;
+ when(external.getExecutor()).thenReturn(completionExecutor);
+ when(v4.closeAsync()).thenAnswer(__ -> {
+ completions.shutdownNow();
+ dispatchExecutor.shutdownNow();
+ return CompletableFuture.completedFuture(null);
+ });
+ create(L0);
+ client.registerProducer(producer);
+
+ send(KEY_A, "a1");
+ applyLayout(L1, L0);
+ // Held behind a1, which is in flight on the retiring segment 0.
+ var a2 = send(KEY_A, "a2");
+ assertThat(producer.waitingSendCount()).isEqualTo(1);
+
+ assertThat(client.closeAsync()).isCompleted();
+ assertClosedFailure(a2);
+ }
+
+ @Test
+ public void closingTheClientWaitsForAProducerThatIsAlreadyClosing() {
+ create(L0);
+ client.registerProducer(producer);
+ var a1 = send(KEY_A, "a1");
+ assertThat(values(0)).containsExactly("a1");
+
+ // The segment producer takes its time to close.
+ var segmentClosed = new CompletableFuture<Void>();
+ segmentCloseResult = segmentClosed;
+ var producerClosed = producer.closeAsync();
+ assertThat(producerClosed).isNotDone();
+
+ var clientClosed = client.closeAsync();
+ assertThat(clientClosed).isNotDone();
+ verify(v4, never()).closeAsync();
+
+ complete(0, new AlreadyClosedException("closed"));
+ segmentClosed.complete(null);
+ assertThat(producerClosed).isCompleted();
+ assertThat(clientClosed).isCompleted();
+ verify(v4).closeAsync();
+ assertThat(a1).isCompletedExceptionally();
+ assertThat(memory.currentUsage()).isZero();
+ }
+
+ @Test
+ public void closingTheClientDoesNotWaitForASegmentProducerThatIsNotCreatedYet() {
+ // The segment producer waits for exclusive access, which another client holds.
+ var creation = new CompletableFuture<Producer<String>>();
+ doAnswer(__ -> creation).when(v4).createSegmentProducerAsync(any(), any());
+ create(L0);
+ client.registerProducer(producer);
+ var a1 = send(KEY_A, "a1");
+ assertThat(a1).isNotDone();
+
+ assertThat(client.closeAsync()).isCompleted();
+ verify(v4).closeAsync();
+ assertClosedFailure(a1);
+ assertThat(memory.currentUsage()).isZero();
+
+ // Created after all: it is closed, and nothing is sent to it.
+ creation.complete(segmentProducer(0));
+ dispatchExecutor.runAll();
+ assertThat(closedSegments).contains(0L);
+ assertThat(sent).isEmpty();
+ }
+
+ private static void assertClosedFailure(CompletableFuture<org.apache.pulsar.client.api.v5.MessageId> send) {
+ Class<?> alreadyClosed = org.apache.pulsar.client.api.v5.PulsarClientException.AlreadyClosedException.class;
+ assertThat(send).failsWithin(0, TimeUnit.SECONDS).withThrowableThat()
+ .havingCause().isInstanceOf(alreadyClosed);
+ }
+
+ private void create(ClientSegmentLayout initial) {
+ DagWatchClient watch = mock(DagWatchClient.class);
+ when(watch.topicName()).thenReturn(TOPIC);
+ doAnswer(invocation -> {
+ listener = invocation.getArgument(0);
+ return null;
+ }).when(watch).setListener(any());
+ ProducerConfigurationData conf = new ProducerConfigurationData();
+ conf.setTopicName(TOPIC.toString());
+ dispatchExecutor.testThread = Thread.currentThread();
+ producer = new ScalableTopicProducer<>(client, Schema.string(), conf, watch, initial);
+ }
+
+ private CompletableFuture<org.apache.pulsar.client.api.v5.MessageId> send(String key, String value) {
+ var future = producer.sendInternalAsync(key, value, null, null, null, null, null, null, null);
+ dispatchExecutor.runAll();
+ return future;
+ }
+
+ private void applyLayout(ClientSegmentLayout layout, ClientSegmentLayout previous) {
+ listener.onLayoutChange(layout, previous);
+ dispatchExecutor.runAll();
+ }
+
+ /** Deliver the result of the sends pending on a segment: acknowledge them, or fail them. */
+ private void complete(long segmentId, Throwable failure) {
+ complete(s -> s.segmentId() == segmentId, failure);
+ }
+
+ private void completeAll() {
+ complete(__ -> true, null);
+ }
+
+ private void complete(Predicate<Sent> which, Throwable failure) {
+ List<Sent> pending;
+ synchronized (sent) {
+ pending = sent.stream().filter(s -> !s.result().isDone() && which.test(s)).toList();
+ }
+ for (Sent s : pending) {
+ if (failure == null) {
+ s.result().complete(MessageId.earliest);
+ } else {
+ s.result().completeExceptionally(failure);
+ }
+ }
+ dispatchExecutor.runAll();
+ }
+
+ /** The values handed to a segment's producer, in the order it was given them. */
+ private List<String> values(long segmentId) {
+ return valuesOfKey(segmentId, null);
+ }
+
+ private List<String> valuesOfKey(long segmentId, String key) {
+ synchronized (sent) {
+ return sent.stream()
+ .filter(s -> s.segmentId() == segmentId && (key == null || key.equals(s.key())))
+ .map(Sent::value)
+ .toList();
+ }
+ }
+
+ private static long segmentOf(CompletableFuture<org.apache.pulsar.client.api.v5.MessageId> send) {
+ return ((MessageIdV5) send.join()).segmentId();
+ }
+
+ /** A segment producer that records the messages it is given, for the test to complete. */
+ @SuppressWarnings("unchecked")
+ private Producer<String> segmentProducer(long segmentId) {
+ Producer<String> producer = mock(Producer.class);
+ when(producer.newMessage()).thenAnswer(__ -> {
+ TypedMessageBuilderImpl<String> builder = mock(TypedMessageBuilderImpl.class, RETURNS_SELF);
+ String[] key = new String[1];
+ String[] value = new String[1];
+ when(builder.key(anyString())).thenAnswer(invocation -> {
+ key[0] = invocation.getArgument(0);
+ return builder;
+ });
+ when(builder.encodedValue(any())).thenAnswer(invocation -> {
+ value[0] = new String(((EncodeData) invocation.getArgument(0)).data(), StandardCharsets.UTF_8);
+ return builder;
+ });
+ when(builder.sendAsync()).thenAnswer(invocation -> {
+ if (closedSegments.contains(segmentId)) {
+ return CompletableFuture.failedFuture(new AlreadyClosedException("closed"));
+ }
+ CompletableFuture<MessageId> result = new CompletableFuture<>();
+ sent.add(new Sent(segmentId, key[0], value[0], result));
+ return result;
+ });
+ return builder;
+ });
+ when(producer.closeAsync()).thenAnswer(__ -> {
+ closedSegments.add(segmentId);
+ return segmentCloseResult;
+ });
+ return producer;
+ }
+
+ /** A key whose segment-routing hash is in {@code [start, end]}. */
+ private static String keyIn(int start, int end) {
+ for (int i = 0; ; i++) {
+ String key = "key-" + i;
+ int hash = SegmentRouter.hash(key);
+ if (hash >= start && hash <= end) {
+ return key;
+ }
+ }
+ }
+
+ private static ClientSegmentLayout layout(long epoch) {
+ ScalableTopicDAG dag = new ScalableTopicDAG().setEpoch(epoch);
+ 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(epoch >= 2 ? SegmentState.SEALED : SegmentState.ACTIVE).setCreatedAtEpoch(1)
+ .addParentId(0);
+ dag.addSegment().setSegmentId(2).setHashStart(0x8000).setHashEnd(0xffff)
+ .setState(SegmentState.ACTIVE).setCreatedAtEpoch(1).addParentId(0);
+ }
+ if (epoch >= 2) {
+ dag.addSegment().setSegmentId(3).setHashStart(0).setHashEnd(0x3fff)
+ .setState(SegmentState.ACTIVE).setCreatedAtEpoch(2).addParentId(1);
+ dag.addSegment().setSegmentId(4).setHashStart(0x4000).setHashEnd(0x7fff)
+ .setState(SegmentState.ACTIVE).setCreatedAtEpoch(2).addParentId(1);
+ }
+ return ClientSegmentLayout.fromProto(dag, TOPIC);
+ }
+
+ /**
+ * The dispatch executor: runs its tasks in order on the test thread, one at a time, as the
+ * client's single dispatch thread would. A task submitted on the test thread runs right away,
+ * unless another is running, or the executor is paused, as one that is behind; then it waits for
+ * the next {@link #runAll}. Tasks from other threads, such as the retirement check's timer, wait
+ * for it too.
+ */
+ private static final class TestExecutor extends AbstractExecutorService {
+ /** The thread running the test: the one that created the producer. */
+ volatile Thread testThread;
+ final ConcurrentLinkedQueue<Runnable> tasks = new ConcurrentLinkedQueue<>();
+ private boolean running;
+ volatile boolean paused;
+ private volatile boolean shutdown;
+
+ @Override
+ public void execute(Runnable task) {
+ if (shutdown) {
+ throw new RejectedExecutionException("shut down");
+ }
+ tasks.add(task);
+ runAll();
+ }
+
+ void runAll() {
+ if (running || paused || Thread.currentThread() != testThread) {
+ return;
+ }
+ running = true;
+ try {
+ Runnable task;
+ while ((task = tasks.poll()) != null) {
+ task.run();
+ }
+ } finally {
+ running = false;
+ }
+ }
+
+ @Override
+ public List<Runnable> shutdownNow() {
+ shutdown = true;
+ List<Runnable> dropped = new ArrayList<>(tasks);
+ tasks.clear();
+ return dropped;
+ }
+
+ @Override
+ public void shutdown() {
+ shutdown = true;
+ }
+
+ @Override
+ public boolean isShutdown() {
+ return shutdown;
+ }
+
+ @Override
+ public boolean isTerminated() {
+ return shutdown && tasks.isEmpty();
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit) {
+ return isTerminated();
+ }
+ }
+}