[Java Client] Send CloseProducer on timeout (#13161)

* [Java Client] Send CloseProducer on timeout * Fix failed test from ClientErrorsTest (cherry picked from commit 27d5429)
apache · Dec 11, 2021 · 50632a7 · 50632a7
1 parent 32f7b4f
commit 50632a7
Show file tree

Hide file tree

Showing 2 changed files with 68 additions and 4 deletions.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java
@@ -31,12 +31,14 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
+import lombok.Cleanup;
 import org.apache.pulsar.client.impl.ConsumerBase;
 import org.apache.pulsar.client.impl.ProducerBase;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandLookupTopicResponse.LookupType;
 import org.apache.pulsar.common.api.proto.PulsarApi.ServerError;
 import org.apache.pulsar.common.protocol.Commands;
 import org.apache.pulsar.common.protocol.schema.SchemaVersion;
+import org.awaitility.Awaitility;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
@@ -167,6 +169,7 @@ private void producerCreateFailAfterRetryTimeout(String topic) throws Exception
         PulsarClient client = PulsarClient.builder().serviceUrl(mockBrokerService.getBrokerAddress())
                 .operationTimeout(1, TimeUnit.SECONDS).build();
         final AtomicInteger counter = new AtomicInteger(0);
+        final AtomicInteger closeProducerCounter = new AtomicInteger(0);
 
         mockBrokerService.setHandleProducer((ctx, producer) -> {
             if (counter.incrementAndGet() == 2) {
@@ -179,16 +182,69 @@ private void producerCreateFailAfterRetryTimeout(String topic) throws Exception
             ctx.writeAndFlush(Commands.newError(producer.getRequestId(), ServerError.ServiceNotReady, "msg"));
         });
 
+        mockBrokerService.setHandleCloseProducer((ctx, closeProducer) -> {
+            closeProducerCounter.incrementAndGet();
+        });
+
         try {
             client.newProducer().topic(topic).create();
             fail("Should have failed");
         } catch (Exception e) {
             // we fail even on the retriable error
             assertTrue(e instanceof PulsarClientException);
         }
+        // There is a small race condition here because the producer's timeout both fails the client creation
+        // and triggers sending CloseProducer.
+        Awaitility.await().until(() -> closeProducerCounter.get() == 1);
+        mockBrokerService.resetHandleProducer();
+        mockBrokerService.resetHandleCloseProducer();
+    }
+
+    @Test
+    public void testCreatedProducerSendsCloseProducerAfterTimeout() throws Exception {
+        producerCreatedThenFailsRetryTimeout("persistent://prop/use/ns/t1");
+    }
+
+    @Test
+    public void testCreatedPartitionedProducerSendsCloseProducerAfterTimeout() throws Exception {
+        producerCreatedThenFailsRetryTimeout("persistent://prop/use/ns/part-t1");
+    }
+
+    private void producerCreatedThenFailsRetryTimeout(String topic) throws Exception {
+        @Cleanup
+        PulsarClient client = PulsarClient.builder().serviceUrl(mockBrokerService.getBrokerAddress())
+                .operationTimeout(1, TimeUnit.SECONDS).build();
+        final AtomicInteger producerCounter = new AtomicInteger(0);
+        final AtomicInteger closeProducerCounter = new AtomicInteger(0);
+
+        mockBrokerService.setHandleProducer((ctx, producer) -> {
+            int producerCount = producerCounter.incrementAndGet();
+            if (producerCount == 1) {
+                ctx.writeAndFlush(Commands.newProducerSuccess(producer.getRequestId(), "producer1",
+                        SchemaVersion.Empty));
+                // Trigger reconnect
+                ctx.writeAndFlush(Commands.newCloseProducer(producer.getProducerId(), -1));
+            } else if (producerCount != 2) {
+                // Respond to subsequent requests to prevent timeouts
+                ctx.writeAndFlush(Commands.newProducerSuccess(producer.getRequestId(), "producer1",
+                        SchemaVersion.Empty));
+            }
+            // Don't respond to the second Producer command to ensure timeout
+        });
 
+        mockBrokerService.setHandleCloseProducer((ctx, closeProducer) -> {
+            closeProducerCounter.incrementAndGet();
+            ctx.writeAndFlush(Commands.newSuccess(closeProducer.getRequestId()));
+        });
+
+        // Create producer should succeed then upon closure, it should reattempt creation. The first request will
+        // timeout, which triggers CloseProducer. The client might send send the third Producer command before the
+        // below assertion, so we pass with 2 or 3.
+        client.newProducer().topic(topic).create();
+        Awaitility.await().until(() -> closeProducerCounter.get() == 1);
+        Awaitility.await().until(() -> producerCounter.get() == 2 || producerCounter.get() == 3);
         mockBrokerService.resetHandleProducer();
-        client.close();
+        mockBrokerService.resetHandleCloseProducer();
     }
 
     @Test
@@ -461,8 +517,6 @@ private void subscribeFailDoesNotFailOtherConsumer(String topic1, String topic2)
         client.close();
     }
 
-    // if a producer fails to connect while creating partitioned producer, it should close all successful connections of
-    // other producers and fail
     @Test
     public void testOneProducerFailShouldCloseAllProducersInPartitionedProducer() throws Exception {
         PulsarClient client = PulsarClient.builder().serviceUrl(mockBrokerService.getHttpAddress()).build();

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -1356,7 +1356,17 @@ public void connectionOpened(final ClientCnx cnx) {
                         return null;
                     }
                     log.error("[{}] [{}] Failed to create producer: {}", topic, producerName, cause.getMessage());
-                    // Close the producer since topic does not exists.
+
+                    if (cause instanceof TimeoutException) {
+                        // Creating the producer has timed out. We need to ensure the broker closes the producer
+                        // in case it was indeed created, otherwise it might prevent new create producer operation,
+                        // since we are not necessarily closing the connection.
+                        long closeRequestId = client.newRequestId();
+                        ByteBuf cmd = Commands.newCloseProducer(producerId, closeRequestId);
+                        cnx.sendRequestWithId(cmd, closeRequestId);
+                    }
+
+                    // Close the producer since topic does not exist.
                     if (cause instanceof PulsarClientException.TopicDoesNotExistException) {
                         closeAsync().whenComplete((v, ex) -> {
                             if (ex != null) {