(pulsar) branch master updated: [fix][client] Complete a v5 send in place while its client is closing (#26686)

[mm...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

merlimat 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 8d90feaddb52 [fix][client] Complete a v5 send in place while its client is closing (#26686)
8d90feaddb52 is described below

commit 8d90feaddb529533c97b9081381919cdd12c7ee7
Author: SeungTaek Lim <na...@gmail.com>
AuthorDate: Wed Sep 23 07:08:39 2026 +0900

    [fix][client] Complete a v5 send in place while its client is closing (#26686)
---
 .../impl/v5/V5ProducerClosingCompletionTest.java   | 106 +++++++++++++++++++++
 .../client/impl/v5/ScalableTopicProducer.java      |   6 +-
 2 files changed, 111 insertions(+), 1 deletion(-)

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerClosingCompletionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerClosingCompletionTest.java
new file mode 100644
index 000000000000..31b12d450cfe
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5ProducerClosingCompletionTest.java
@@ -0,0 +1,106 @@
+/*
+ * 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.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.client.api.v5.MessageId;
+import org.apache.pulsar.client.api.v5.Producer;
+import org.apache.pulsar.client.api.v5.PulsarClient;
+import org.apache.pulsar.client.api.v5.PulsarClientException;
+import org.apache.pulsar.client.api.v5.V5ClientBaseTest;
+import org.apache.pulsar.client.api.v5.config.BatchingPolicy;
+import org.apache.pulsar.client.api.v5.config.ConnectionPolicy;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+/**
+ * A send that fails because its client is closing must still complete its future. The failure is
+ * delivered on the client's completion executor, which the client shuts down with
+ * {@code shutdownNow()}, dropping whatever is still queued on it at that moment.
+ */
+public class V5ProducerClosingCompletionTest extends V5ClientBaseTest {
+
+    /** One callback thread, so that holding it holds the producer's only completion executor. */
+    private PulsarClient newClientWithOneCallbackThread() throws Exception {
+        return track(PulsarClient.builder()
+                .serviceUrl(getBrokerServiceUrl())
+                .connectionPolicy(ConnectionPolicy.builder().ioThreads(1).callbackThreads(1).build())
+                .build());
+    }
+
+    private static void awaitQuietly(CountDownLatch gate) {
+        try {
+            gate.await();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+
+    @Test
+    public void aSendFailedWhileTheClientIsClosingCompletesEvenIfItsCompletionWasQueued() throws Exception {
+        PulsarClient client = newClientWithOneCallbackThread();
+        PulsarClientImpl v4Client = ((PulsarClientV5) client).v4Client();
+        String topic = newScalableTopic(1);
+        Producer<byte[]> producer = track(client.newProducer(Schema.bytes())
+                .topic(topic)
+                .batchingPolicy(BatchingPolicy.ofDisabled())
+                .create());
+        // The per-segment producer exists, so a later send has nothing to create.
+        producer.newMessage().value(new byte[16]).send();
+
+        // Hold the only completion executor: a completion handed to it now stays queued behind this.
+        CountDownLatch gate = new CountDownLatch(1);
+        v4Client.externalExecutorProvider().getExecutor().execute(() -> awaitQuietly(gate));
+        try {
+            // Close on another thread: close() returns only once the client's executors are gone.
+            CompletableFuture<Void> closing = CompletableFuture.runAsync(() -> {
+                try {
+                    client.close();
+                } catch (Exception e) {
+                    throw new CompletionException(e);
+                }
+            });
+            Awaitility.await().pollDelay(Duration.ZERO).pollInterval(Duration.ofMillis(1))
+                    .until(v4Client::isClosed);
+
+            // Fails right away, the client is closing; its completion is handed to the held executor.
+            CompletableFuture<MessageId> send = producer.async().newMessage().value(new byte[16]).send();
+
+            // shutdownNow(): the held task is interrupted and whatever was queued behind it is dropped.
+            closing.get(30, TimeUnit.SECONDS);
+            gate.countDown();
+
+            ExecutionException failure = expectThrows(ExecutionException.class,
+                    () -> send.get(5, TimeUnit.SECONDS));
+            assertTrue(failure.getCause() instanceof PulsarClientException.AlreadyClosedException,
+                    "the send must fail because the client closed, got " + failure.getCause());
+        } finally {
+            gate.countDown();
+        }
+    }
+}
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 350e2191b5e3..fab665c65d87 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
@@ -676,7 +676,11 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
             return;
         }
         boolean offload = send.dispatching || (failure != null && !isEventLoopThread());
-        if (!offload) {
+        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.
             complete(send, messageId, failure);
             return;
         }