[fix][client] Fix retry logic for BinaryProtoLookupService.getTopicsUnderNamespace method

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java

@@ -48,7 +48,6 @@
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.DefaultChannelId;
 import io.netty.channel.embedded.EmbeddedChannel;
-import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
 import io.vertx.core.impl.ConcurrentHashSet;
 import java.io.Closeable;
 import java.io.IOException;
@@ -154,6 +153,7 @@
 import org.apache.pulsar.common.protocol.ByteBufPair;
 import org.apache.pulsar.common.protocol.Commands;
 import org.apache.pulsar.common.protocol.Commands.ChecksumType;
+import org.apache.pulsar.common.protocol.FrameDecoderUtil;
 import org.apache.pulsar.common.protocol.PulsarHandler;
 import org.apache.pulsar.common.protocol.schema.EmptyVersion;
 import org.apache.pulsar.common.topics.TopicList;
@@ -1212,14 +1212,9 @@ public void close() throws IOException {
     private class ClientChannel implements Closeable {
         private ClientChannelHelper clientChannelHelper = new ClientChannelHelper();
         private ServerCnx serverCnx = new ServerCnx(pulsar);
-        private EmbeddedChannel channel = new EmbeddedChannel(DefaultChannelId.newInstance(),
-                new LengthFieldBasedFrameDecoder(
-                5 * 1024 * 1024,
-                0,
-                4,
-                0,
-                4),
-                serverCnx);
+        private EmbeddedChannel channel =
+                new EmbeddedChannel(DefaultChannelId.newInstance(), FrameDecoderUtil.createFrameDecoder(),
+                        (ChannelHandler) serverCnx);
         public ClientChannel() {
             serverCnx.setAuthRole("");
         }
@@ -2850,20 +2845,13 @@ public void testSendFailureOnEncryptionRequiredTopic() throws Exception {
     }
 
     protected void resetChannel() throws Exception {
-        int maxMessageSize = 5 * 1024 * 1024;
         if (channel != null && channel.isActive()) {
             serverCnx.close();
             channel.close().get();
         }
         serverCnx = new ServerCnx(pulsar);
         serverCnx.setAuthRole("");
-        channel = new EmbeddedChannel(new LengthFieldBasedFrameDecoder(
-                maxMessageSize,
-                0,
-                4,
-                0,
-                4),
-                (ChannelHandler) serverCnx);
+        channel = new EmbeddedChannel(FrameDecoderUtil.createFrameDecoder(), (ChannelHandler) serverCnx);
     }
 
     protected void setChannelConnected() throws Exception {

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/utils/ClientChannelHelper.java

@@ -21,7 +21,6 @@
 import com.google.common.collect.Queues;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.embedded.EmbeddedChannel;
-import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
 import java.util.Queue;
 import org.apache.pulsar.common.api.proto.BaseCommand;
 import org.apache.pulsar.common.api.proto.CommandAck;
@@ -52,6 +51,7 @@
 import org.apache.pulsar.common.api.proto.CommandSuccess;
 import org.apache.pulsar.common.api.proto.CommandUnsubscribe;
 import org.apache.pulsar.common.api.proto.CommandWatchTopicListSuccess;
+import org.apache.pulsar.common.protocol.FrameDecoderUtil;
 import org.apache.pulsar.common.protocol.PulsarDecoder;
 
 public class ClientChannelHelper {
@@ -61,7 +61,7 @@ public class ClientChannelHelper {
 
     public ClientChannelHelper() {
         int maxMessageSize = 5 * 1024 * 1024;
-        channel = new EmbeddedChannel(new LengthFieldBasedFrameDecoder(maxMessageSize, 0, 4, 0, 4), decoder);
+        channel = new EmbeddedChannel(FrameDecoderUtil.createFrameDecoder(), decoder);
     }
 
     public Object getCommand(Object obj) {

pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithInternalListenerNameTest.java

@@ -33,8 +33,8 @@
 import java.util.Collections;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import lombok.Cleanup;
 import org.apache.commons.lang3.tuple.Pair;
@@ -61,7 +61,7 @@
 @Test(groups = "broker-api")
 public class PulsarMultiListenersWithInternalListenerNameTest extends MockedPulsarServiceBaseTest {
     private final boolean withInternalListener;
-    private ExecutorService executorService;
+    private ScheduledExecutorService executorService;
     private InetSocketAddress brokerAddress;
     private InetSocketAddress brokerSslAddress;
     private EventLoopGroup eventExecutors;
@@ -79,7 +79,7 @@ protected PulsarMultiListenersWithInternalListenerNameTest(boolean withInternalL
     @BeforeMethod(alwaysRun = true)
     @Override
     protected void setup() throws Exception {
-        this.executorService = Executors.newFixedThreadPool(1);
+        this.executorService = Executors.newSingleThreadScheduledExecutor();
         this.eventExecutors = new NioEventLoopGroup();
         this.isTcpLookup = true;
         String host = InetAddress.getLocalHost().getHostAddress();

pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java

@@ -34,7 +34,6 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
 import org.apache.commons.lang3.mutable.MutableObject;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.pulsar.client.api.PulsarClientException;
@@ -61,7 +60,7 @@ public class BinaryProtoLookupService implements LookupService {
     private final PulsarClientImpl client;
     private final ServiceNameResolver serviceNameResolver;
     private final boolean useTls;
-    private final ExecutorService scheduleExecutor;
+    private final ScheduledExecutorService scheduleExecutor;
     private final String listenerName;
     private final int maxLookupRedirects;
     private final ExecutorService lookupPinnedExecutor;
@@ -83,27 +82,29 @@ public class BinaryProtoLookupService implements LookupService {
 
     /**
      * @deprecated use {@link
-     * #BinaryProtoLookupService(PulsarClientImpl, String, String, boolean, ExecutorService, ExecutorService)} instead.
+     * #BinaryProtoLookupService(PulsarClientImpl, String, String, boolean, ScheduledExecutorService, ExecutorService)}
+     * instead.
      */
     @Deprecated
     public BinaryProtoLookupService(PulsarClientImpl client,
                                     String serviceUrl,
                                     boolean useTls,
-                                    ExecutorService scheduleExecutor)
+                                    ScheduledExecutorService scheduleExecutor)
             throws PulsarClientException {
         this(client, serviceUrl, null, useTls, scheduleExecutor);
     }
 
     /**
      * @deprecated use {@link
-     * #BinaryProtoLookupService(PulsarClientImpl, String, String, boolean, ExecutorService, ExecutorService)} instead.
+     * #BinaryProtoLookupService(PulsarClientImpl, String, String, boolean, ScheduledExecutorService, ExecutorService)}
+     * instead.
      */
     @Deprecated
     public BinaryProtoLookupService(PulsarClientImpl client,
                                     String serviceUrl,
                                     String listenerName,
                                     boolean useTls,
-                                    ExecutorService scheduleExecutor)
+                                    ScheduledExecutorService scheduleExecutor)
             throws PulsarClientException {
         this(client, serviceUrl, listenerName, useTls, scheduleExecutor, null);
     }
@@ -112,7 +113,7 @@ public BinaryProtoLookupService(PulsarClientImpl client,
                                     String serviceUrl,
                                     String listenerName,
                                     boolean useTls,
-                                    ExecutorService scheduleExecutor,
+                                    ScheduledExecutorService scheduleExecutor,
                                     ExecutorService lookupPinnedExecutor)
             throws PulsarClientException {
         this.client = client;
@@ -409,15 +410,16 @@ public CompletableFuture<GetTopicsResult> getTopicsUnderNamespace(NamespaceName
         try {
             return topicsUnderNamespaceInProgress.computeIfAbsent(key, k -> {
                 CompletableFuture<GetTopicsResult> topicsFuture = new CompletableFuture<>();
-                AtomicLong opTimeoutMs = new AtomicLong(client.getConfiguration().getOperationTimeoutMs());
+                long opTimeoutMs = client.getConfiguration().getOperationTimeoutMs();
                 Backoff backoff = new BackoffBuilder()
                         .setInitialTime(100, TimeUnit.MILLISECONDS)
-                        .setMandatoryStop(opTimeoutMs.get() * 2, TimeUnit.MILLISECONDS)
+                        .setMandatoryStop(opTimeoutMs * 2, TimeUnit.MILLISECONDS)
                         .setMax(1, TimeUnit.MINUTES)
                         .create();
-
+                long startTimeNanos = System.nanoTime();
+                long retryUntilNanos = startTimeNanos + TimeUnit.MILLISECONDS.toNanos(opTimeoutMs);
                 newFutureCreated.setValue(topicsFuture);
-                getTopicsUnderNamespace(namespace, backoff, opTimeoutMs, topicsFuture, mode,
+                getTopicsUnderNamespace(namespace, backoff, startTimeNanos, retryUntilNanos, topicsFuture, mode,
                         topicsPattern, topicsHash);
                 return topicsFuture;
             });
@@ -433,50 +435,56 @@ public CompletableFuture<GetTopicsResult> getTopicsUnderNamespace(NamespaceName
     private void getTopicsUnderNamespace(
                                          NamespaceName namespace,
                                          Backoff backoff,
-                                         AtomicLong remainingTime,
+                                         long startTimeNanos,
+                                         long retryUntilNanos,
                                          CompletableFuture<GetTopicsResult> getTopicsResultFuture,
                                          Mode mode,
                                          String topicsPattern,
                                          String topicsHash) {
-        long startTime = System.nanoTime();
-
-        client.getCnxPool().getConnection(serviceNameResolver).thenAcceptAsync(clientCnx -> {
+        client.getCnxPool().getConnection(serviceNameResolver).thenComposeAsync(clientCnx -> {
             long requestId = client.newRequestId();
-            ByteBuf request = Commands.newGetTopicsOfNamespaceRequest(
-                namespace.toString(), requestId, mode, topicsPattern, topicsHash);
-
-            clientCnx.newGetTopicsOfNamespace(request, requestId).whenComplete((r, t) -> {
-                if (t != null) {
-                    histoListTopics.recordFailure(System.nanoTime() - startTime);
-                    getTopicsResultFuture.completeExceptionally(t);
+            ByteBuf request = Commands.newGetTopicsOfNamespaceRequest(namespace.toString(), requestId, mode,
+                            topicsPattern, topicsHash);
+            return clientCnx.newGetTopicsOfNamespace(request, requestId).whenComplete((r, t) -> {
+                client.getCnxPool().releaseConnection(clientCnx);
+            });
+        }, lookupPinnedExecutor).whenComplete((r, t) -> {
+            if (t != null) {
+                Throwable cause = FutureUtil.unwrapCompletionException(t);
+                if (cause instanceof PulsarClientException && !PulsarClientException.isRetriableError(cause)) {
+                    histoListTopics.recordFailure(System.nanoTime() - startTimeNanos);
+                    getTopicsResultFuture.completeExceptionally(cause);
+                    return;
+                }
+                long nowNanos = System.nanoTime();
+                if (nowNanos > retryUntilNanos) {
+                    histoListTopics.recordFailure(System.nanoTime() - startTimeNanos);
+                    log.warn("[namespace: {}] Error while getTopicsUnderNamespace -- No more retries left."
+                            + " Last error was {}", namespace, t.getMessage());
+                    getTopicsResultFuture.completeExceptionally(new PulsarClientException.TimeoutException(
+                            format("Could not get topics of namespace %s within configured timeout",
+                                    namespace.toString())));
                 } else {
-                    histoListTopics.recordSuccess(System.nanoTime() - startTime);
-                    if (log.isDebugEnabled()) {
-                        log.debug("[namespace: {}] Success get topics list in request: {}",
-                                namespace, requestId);
+                    long nextDelay = backoff.next();
+                    log.warn("[namespace: {}] Error while getTopicsUnderNamespace -- Will try again in"
+                            + " {} ms. Error was {}", namespace, nextDelay, t.getMessage());
+                    if (!getTopicsResultFuture.isDone()) {
+                        scheduleExecutor.schedule(() -> {
+                            getTopicsUnderNamespace(namespace, backoff, startTimeNanos, retryUntilNanos,
+                                    getTopicsResultFuture, mode, topicsPattern, topicsHash);
+                        }, nextDelay, TimeUnit.MILLISECONDS);
+                    } else {
+                        log.info("[namespace: {}] Ignoring retry in getTopicsUnderNamespace -- Future is already "
+                                + "completed", namespace);
                     }
-                    getTopicsResultFuture.complete(r);
                 }
-                client.getCnxPool().releaseConnection(clientCnx);
-            });
-        }, lookupPinnedExecutor).exceptionally((e) -> {
-            long nextDelay = Math.min(backoff.next(), remainingTime.get());
-            if (nextDelay <= 0) {
-                getTopicsResultFuture.completeExceptionally(
-                    new PulsarClientException.TimeoutException(
-                        format("Could not get topics of namespace %s within configured timeout",
-                            namespace.toString())));
-                return null;
+            } else {
+                histoListTopics.recordSuccess(System.nanoTime() - startTimeNanos);
+                if (log.isDebugEnabled()) {
+                    log.debug("[namespace: {}] Success get topics list", namespace);
+                }
+                getTopicsResultFuture.complete(r);
             }
-
-            ((ScheduledExecutorService) scheduleExecutor).schedule(() -> {
-                log.warn("[namespace: {}] Could not get connection while getTopicsUnderNamespace -- Will try again in"
-                                + " {} ms", namespace, nextDelay);
-                remainingTime.addAndGet(-nextDelay);
-                getTopicsUnderNamespace(namespace, backoff, remainingTime, getTopicsResultFuture,
-                        mode, topicsPattern, topicsHash);
-            }, nextDelay, TimeUnit.MILLISECONDS);
-            return null;
         });
     }
 

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java

@@ -27,7 +27,6 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.EventLoopGroup;
 import io.netty.channel.unix.Errors.NativeIoException;
 import io.netty.handler.ssl.SslHandshakeCompletionEvent;
 import io.netty.util.concurrent.Promise;
@@ -43,6 +42,7 @@
 import java.util.Queue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
@@ -166,7 +166,7 @@ public class ClientCnx extends PulsarHandler {
     @Getter(AccessLevel.PACKAGE)
     private final Semaphore pendingLookupRequestSemaphore;
     private final Semaphore maxLookupRequestSemaphore;
-    private final EventLoopGroup eventLoopGroup;
+    private final ScheduledExecutorService scheduledExecutor;
 
     private static final AtomicIntegerFieldUpdater<ClientCnx> NUMBER_OF_REJECTED_REQUESTS_UPDATER =
             AtomicIntegerFieldUpdater.newUpdater(ClientCnx.class, "numberOfRejectRequests");
@@ -251,20 +251,20 @@ String getDescription() {
     }
 
     public ClientCnx(InstrumentProvider instrumentProvider,
-                     ClientConfigurationData conf, EventLoopGroup eventLoopGroup) {
-        this(instrumentProvider, conf, eventLoopGroup, Commands.getCurrentProtocolVersion());
+                     ClientConfigurationData conf, ScheduledExecutorService scheduledExecutor) {
+        this(instrumentProvider, conf, scheduledExecutor, Commands.getCurrentProtocolVersion());
     }
 
-    public ClientCnx(InstrumentProvider instrumentProvider, ClientConfigurationData conf, EventLoopGroup eventLoopGroup,
-                     int protocolVersion) {
+    public ClientCnx(InstrumentProvider instrumentProvider, ClientConfigurationData conf,
+                     ScheduledExecutorService scheduledExecutor, int protocolVersion) {
         super(conf.getKeepAliveIntervalSeconds(), TimeUnit.SECONDS);
         checkArgument(conf.getMaxLookupRequest() > conf.getConcurrentLookupRequest());
         this.pendingLookupRequestSemaphore = new Semaphore(conf.getConcurrentLookupRequest(), false);
         this.maxLookupRequestSemaphore =
                 new Semaphore(conf.getMaxLookupRequest() - conf.getConcurrentLookupRequest(), false);
         this.waitingLookupRequests = Queues.newConcurrentLinkedQueue();
         this.authentication = conf.getAuthentication();
-        this.eventLoopGroup = eventLoopGroup;
+        this.scheduledExecutor = scheduledExecutor;
         this.maxNumberOfRejectedRequestPerConnection = conf.getMaxNumberOfRejectedRequestPerConnection();
         this.operationTimeoutMs = conf.getOperationTimeoutMs();
         this.state = State.None;
@@ -289,7 +289,7 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
         this.localAddress = ctx.channel().localAddress();
         this.remoteAddress = ctx.channel().remoteAddress();
 
-        this.timeoutTask = this.eventLoopGroup
+        this.timeoutTask = this.scheduledExecutor
                 .scheduleAtFixedRate(catchingAndLoggingThrowables(this::checkRequestTimeout), operationTimeoutMs,
                         operationTimeoutMs, TimeUnit.MILLISECONDS);
 
@@ -759,7 +759,7 @@ private CompletableFuture<LookupDataResult> getAndRemovePendingLookupRequest(lon
             if (firstOneWaiting != null) {
                 maxLookupRequestSemaphore.release();
                 // schedule a new lookup in.
-                eventLoopGroup.execute(() -> {
+                scheduledExecutor.execute(() -> {
                     long newId = firstOneWaiting.getLeft();
                     TimedCompletableFuture<LookupDataResult> newFuture = firstOneWaiting.getRight().getRight();
                     addPendingLookupRequests(newId, newFuture);
@@ -1291,7 +1291,7 @@ private void incrementRejectsAndMaybeClose() {
         long rejectedRequests = NUMBER_OF_REJECTED_REQUESTS_UPDATER.getAndIncrement(this);
         if (rejectedRequests == 0) {
             // schedule timer
-            eventLoopGroup.schedule(() -> NUMBER_OF_REJECTED_REQUESTS_UPDATER.set(ClientCnx.this, 0),
+            scheduledExecutor.schedule(() -> NUMBER_OF_REJECTED_REQUESTS_UPDATER.set(ClientCnx.this, 0),
                                     rejectedRequestResetTimeSec, TimeUnit.SECONDS);
         } else if (rejectedRequests >= maxNumberOfRejectedRequestPerConnection) {
             log.error("{} Close connection because received {} rejected request in {} seconds ", ctx.channel(),