[GitHub] rdhabalia closed pull request #3240: timeout stuck producer/consumer operation request

[GitBox <gi...@apache.org>]

rdhabalia closed pull request #3240: timeout stuck producer/consumer operation request
URL: https://github.com/apache/pulsar/pull/3240
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

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
index a0f007a839..def895b631 100644
--- 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
@@ -189,7 +189,7 @@ private void producerCreateFailAfterRetryTimeout(String topic) throws Exception
             fail("Should have failed");
         } catch (Exception e) {
             // we fail even on the retriable error
-            assertTrue(e instanceof PulsarClientException.LookupException);
+            assertTrue(e instanceof PulsarClientException);
         }
 
         mockBrokerService.resetHandleProducer();
@@ -417,7 +417,7 @@ private void subscribeFailAfterRetryTimeout(String topic) throws Exception {
             fail("Should have failed");
         } catch (Exception e) {
             // we fail even on the retriable error
-            assertEquals(e.getClass(), LookupException.class);
+            assertTrue(e instanceof PulsarClientException);
         }
 
         mockBrokerService.resetHandleSubscribe();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
index 314ca699c1..6988863eb4 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
@@ -38,6 +38,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.NavigableMap;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentSkipListMap;
@@ -69,6 +70,7 @@
 import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
 import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
 import org.apache.pulsar.zookeeper.ZooKeeperDataCache;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -88,7 +90,7 @@
     @Override
     protected void setup() throws Exception {
         super.internalSetup();
-        producerBaseSetup();
+        super.producerBaseSetup();
     }
 
     @AfterMethod
@@ -732,4 +734,27 @@ public void testCleanProducer() throws Exception {
         log.info("-- Exiting {} test --", methodName);
     }
 
+    /**
+     * It verifies that if broker fails to complete producer/consumer operation then client times out rather waiting
+     * forever.
+     * 
+     * @throws PulsarClientException
+     */
+    @Test(expectedExceptions = PulsarClientException.TimeoutException.class, timeOut = 10000)
+    public void testOperationTimeout() throws PulsarClientException {
+        final String topicName = "persistent://my-property/my-ns/my-topic1";
+        ConcurrentOpenHashMap<String, CompletableFuture<Optional<Topic>>> topics = pulsar.getBrokerService()
+                .getTopics();
+        // non-complete topic future so, create topic should timeout
+        topics.put(topicName, new CompletableFuture<>());
+        PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(lookupUrl.toString())
+                .operationTimeout(2, TimeUnit.SECONDS).statsInterval(0, TimeUnit.SECONDS).build();
+        try {
+            Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
+        } finally {
+            topics.clear();
+            pulsarClient.close();
+        }
+    }
+
 }
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
index ccc20b3ec3..84d2461658 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
@@ -21,18 +21,6 @@
 import static com.google.common.base.Preconditions.checkArgument;
 import static org.apache.pulsar.client.impl.HttpClient.getPulsarClientVersion;
 
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.Queues;
-
-import io.netty.buffer.ByteBuf;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelHandler;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.EventLoopGroup;
-import io.netty.channel.unix.Errors.NativeIoException;
-import io.netty.handler.ssl.SslHandler;
-import io.netty.util.concurrent.Promise;
-
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.nio.channels.ClosedChannelException;
@@ -40,6 +28,7 @@
 import java.util.Optional;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
@@ -78,6 +67,18 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Queues;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.unix.Errors.NativeIoException;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.Promise;
+
 public class ClientCnx extends PulsarHandler {
 
     protected final Authentication authentication;
@@ -101,6 +102,7 @@
     private final ConcurrentLongHashMap<ConsumerImpl<?>> consumers = new ConcurrentLongHashMap<>(16, 1);
 
     private final CompletableFuture<Void> connectionFuture = new CompletableFuture<Void>();
+    private final ConcurrentLinkedQueue<RequestTime> requestTimeoutQueue = new ConcurrentLinkedQueue<>();
     private final Semaphore pendingLookupRequestSemaphore;
     private final EventLoopGroup eventLoopGroup;
 
@@ -123,6 +125,17 @@
         None, SentConnectFrame, Ready, Failed
     }
 
+    static class RequestTime {
+        long creationTimeMs;
+        long requestId;
+
+        public RequestTime(long creationTime, long requestId) {
+            super();
+            this.creationTimeMs = creationTime;
+            this.requestId = requestId;
+        }
+    }
+    
     public ClientCnx(ClientConfigurationData conf, EventLoopGroup eventLoopGroup) {
         this(conf, eventLoopGroup, Commands.getCurrentProtocolVersion());
     }
@@ -141,6 +154,7 @@ public ClientCnx(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, in
         this.isTlsHostnameVerificationEnable = conf.isTlsHostnameVerificationEnable();
         this.hostnameVerifier = new DefaultHostnameVerifier();
         this.protocolVersion = protocolVersion;
+        this.eventLoopGroup.scheduleAtFixedRate(() -> checkRequestTimeout(), operationTimeoutMs, operationTimeoutMs, TimeUnit.MILLISECONDS);
     }
 
     @Override
@@ -668,6 +682,7 @@ SocketAddress serverAddrees() {
                 future.completeExceptionally(writeFuture.cause());
             }
         });
+        requestTimeoutQueue.add(new RequestTime(System.currentTimeMillis(), requestId));
         return future;
     }
 
@@ -828,5 +843,24 @@ public void close() {
        }
     }
 
+    private void checkRequestTimeout() {
+        while (!requestTimeoutQueue.isEmpty()) {
+            RequestTime request = requestTimeoutQueue.peek();
+            if (request == null || (System.currentTimeMillis() - request.creationTimeMs) < operationTimeoutMs) {
+                // if there is no request that is timed out then exit the loop
+                break;
+            }
+            request = requestTimeoutQueue.poll();
+            CompletableFuture<ProducerResponse> requestFuture = pendingRequests.remove(request.requestId);
+            if (requestFuture != null && !requestFuture.isDone()
+                    && requestFuture.completeExceptionally(new TimeoutException(
+                            request.requestId + " lookup request timedout after ms " + operationTimeoutMs))) {
+                log.warn("{} request {} timed out after {} ms", ctx.channel(), request.requestId, operationTimeoutMs);
+            } else {
+                // request is already completed successfully.
+            }
+        }
+    }
+    
     private static final Logger log = LoggerFactory.getLogger(ClientCnx.class);
 }


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services