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/20 22:21:30 UTC
(pulsar) 09/21: [fix][client] Fix retry logic for BinaryProtoLookupService.getTopicsUnderNamespace method (#24974)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 57b2b4f75be25030cfbd388e236bfabf0e23996b
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sat Oct 17 07:42:04 2026 -0700
[fix][client] Fix retry logic for BinaryProtoLookupService.getTopicsUnderNamespace method (#24974)
Co-authored-by: Matteo Merli <mm...@apache.org>
(cherry picked from commit ae1a9be736e4d5dd381c0a4bfc1fe7688d1879b3)
---
...MultiListenersWithInternalListenerNameTest.java | 6 +-
.../client/impl/BinaryProtoLookupService.java | 102 ++++++------
.../pulsar/client/impl/PulsarClientImpl.java | 3 +-
.../client/impl/BinaryProtoLookupServiceTest.java | 172 ++++++++++++++++++++-
4 files changed, 229 insertions(+), 54 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithInternalListenerNameTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithInternalListenerNameTest.java
index 094c32c7ae69..e1f70ad37c6e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithInternalListenerNameTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PulsarMultiListenersWithInternalListenerNameTest.java
@@ -31,8 +31,8 @@ import java.time.Duration;
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;
@@ -59,7 +59,7 @@ import org.testng.annotations.Test;
@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;
@@ -77,7 +77,7 @@ public class PulsarMultiListenersWithInternalListenerNameTest extends MockedPuls
@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();
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java
index c91a474eb875..e2050951bd42 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java
@@ -30,9 +30,9 @@ import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SchemaSerializationException;
import org.apache.pulsar.client.impl.metrics.LatencyHistogram;
@@ -57,7 +57,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;
@@ -69,27 +69,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);
}
@@ -98,7 +100,7 @@ public class BinaryProtoLookupService implements LookupService {
String serviceUrl,
String listenerName,
boolean useTls,
- ExecutorService scheduleExecutor,
+ ScheduledExecutorService scheduleExecutor,
ExecutorService lookupPinnedExecutor)
throws PulsarClientException {
this.client = client;
@@ -355,13 +357,14 @@ public class BinaryProtoLookupService implements LookupService {
String topicsPattern,
String topicsHash) {
CompletableFuture<GetTopicsResult> topicsFuture = new CompletableFuture<>();
- AtomicLong opTimeoutMs = new AtomicLong(client.getConfiguration().getOperationTimeoutMs());
+ long startTimeNanos = System.nanoTime();
+ long retryUntilNanos =
+ startTimeNanos + TimeUnit.MILLISECONDS.toNanos(client.getConfiguration().getOperationTimeoutMs());
Backoff backoff = new BackoffBuilder()
.setInitialTime(100, TimeUnit.MILLISECONDS)
- .setMandatoryStop(opTimeoutMs.get() * 2, TimeUnit.MILLISECONDS)
.setMax(1, TimeUnit.MINUTES)
.create();
- getTopicsUnderNamespace(serviceNameResolver.resolveHost(), namespace, backoff, opTimeoutMs,
+ getTopicsUnderNamespace(serviceNameResolver.resolveHost(), namespace, backoff, startTimeNanos, retryUntilNanos,
topicsFuture, mode, topicsPattern, topicsHash);
return topicsFuture;
}
@@ -374,50 +377,59 @@ public class BinaryProtoLookupService implements LookupService {
private void getTopicsUnderNamespace(InetSocketAddress socketAddress,
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(socketAddress).thenAcceptAsync(clientCnx -> {
+ client.getCnxPool().getConnection(socketAddress).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);
- } else {
- histoListTopics.recordSuccess(System.nanoTime() - startTime);
- if (log.isDebugEnabled()) {
- log.debug("[namespace: {}] Success get topics list in request: {}",
- namespace, requestId);
- }
- getTopicsResultFuture.complete(r);
+ return clientCnx.newGetTopicsOfNamespace(request, requestId)
+ .whenComplete((r, t) -> client.getCnxPool().releaseConnection(clientCnx));
+ }, lookupPinnedExecutor).whenComplete((r, t) -> {
+ if (t == null) {
+ histoListTopics.recordSuccess(System.nanoTime() - startTimeNanos);
+ if (log.isDebugEnabled()) {
+ log.debug("[namespace: {}] Success get topics list", namespace);
}
- 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;
+ getTopicsResultFuture.complete(r);
+ return;
+ }
+ Throwable cause = FutureUtil.unwrapCompletionException(t);
+ if (cause instanceof PulsarClientException && !PulsarClientException.isRetriableError(cause)) {
+ histoListTopics.recordFailure(System.nanoTime() - startTimeNanos);
+ getTopicsResultFuture.completeExceptionally(cause);
+ return;
+ }
+ long remainingNanos = retryUntilNanos - System.nanoTime();
+ if (remainingNanos <= 0) {
+ histoListTopics.recordFailure(System.nanoTime() - startTimeNanos);
+ PulsarClientException.TimeoutException timeoutException =
+ new PulsarClientException.TimeoutException(
+ format("Could not get topics of namespace %s within configured timeout",
+ namespace.toString()));
+ timeoutException.initCause(cause);
+ getTopicsResultFuture.completeExceptionally(timeoutException);
+ return;
+ }
+ if (getTopicsResultFuture.isDone()) {
+ return;
+ }
+ long nextDelayMs = Math.min(backoff.next(), TimeUnit.NANOSECONDS.toMillis(remainingNanos));
+ log.warn("[namespace: {}] Failed to get topics under namespace -- Will try again in {} ms: {}",
+ namespace, nextDelayMs, cause.getMessage());
+ try {
+ scheduleExecutor.schedule(() -> getTopicsUnderNamespace(socketAddress, namespace, backoff,
+ startTimeNanos, retryUntilNanos, getTopicsResultFuture, mode, topicsPattern,
+ topicsHash),
+ nextDelayMs, TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ histoListTopics.recordFailure(System.nanoTime() - startTimeNanos);
+ getTopicsResultFuture.completeExceptionally(cause);
}
-
- ((ScheduledExecutorService) scheduleExecutor).schedule(() -> {
- log.warn("[namespace: {}] Could not get connection while getTopicsUnderNamespace -- Will try again in"
- + " {} ms", namespace, nextDelay);
- remainingTime.addAndGet(-nextDelay);
- getTopicsUnderNamespace(socketAddress, namespace, backoff, remainingTime, getTopicsResultFuture,
- mode, topicsPattern, topicsHash);
- }, nextDelay, TimeUnit.MILLISECONDS);
- return null;
});
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
index e738307e0b00..42f3e52dce70 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
@@ -1265,7 +1265,8 @@ public class PulsarClientImpl implements PulsarClient {
lookupService = new HttpLookupService(instrumentProvider, conf, eventLoopGroup);
} else {
lookupService = new BinaryProtoLookupService(this, url, conf.getListenerName(), conf.isUseTls(),
- this.scheduledExecutorProvider.getExecutor(), this.lookupExecutorProvider.getExecutor());
+ (ScheduledExecutorService) this.scheduledExecutorProvider.getExecutor(),
+ this.lookupExecutorProvider.getExecutor());
}
return new InProgressDeduplicationDecoratorLookupService(lookupService,
() -> getConfiguration().getLookupProperties());
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BinaryProtoLookupServiceTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BinaryProtoLookupServiceTest.java
index cf08fd16392b..619491cfb174 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BinaryProtoLookupServiceTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/BinaryProtoLookupServiceTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.impl;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyLong;
import static org.mockito.Mockito.doAnswer;
@@ -33,30 +34,48 @@ import static org.testng.Assert.assertNotSame;
import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+import com.google.common.util.concurrent.MoreExecutors;
import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.util.concurrent.DefaultThreadFactory;
import java.lang.reflect.Field;
import java.net.InetSocketAddress;
+import java.time.Duration;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.PulsarClientException.LookupException;
import org.apache.pulsar.client.impl.BinaryProtoLookupService.LookupDataResult;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
import org.apache.pulsar.common.api.proto.BaseCommand;
import org.apache.pulsar.common.api.proto.BaseCommand.Type;
+import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace;
import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
+import org.apache.pulsar.common.api.proto.ServerError;
import org.apache.pulsar.common.lookup.GetTopicsResult;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.protocol.Commands;
+import org.apache.pulsar.common.protocol.FrameDecoderUtil;
+import org.apache.pulsar.common.topics.TopicList;
import org.awaitility.Awaitility;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -114,7 +133,7 @@ public class BinaryProtoLookupServiceTest {
doReturn(internalExecutor).when(client).getInternalExecutorService();
lookup = spy(new BinaryProtoLookupService(client, "pulsar://localhost:6650", null, false,
- mock(ExecutorService.class), internalExecutor));
+ mock(ScheduledExecutorService.class), internalExecutor));
topicName = TopicName.get("persistent://tenant1/ns1/t1");
}
@@ -232,7 +251,7 @@ public class BinaryProtoLookupServiceTest {
when(client.instrumentProvider()).thenReturn(InstrumentProvider.NOOP);
when(client.getCnxPool()).thenReturn(cnxPool);
- // Never-completing connection prevents the thenAcceptAsync callback in getTopicsUnderNamespace from executing,
+ // Never-completing connection prevents the thenComposeAsync callback in getTopicsUnderNamespace from executing,
// isolating only the deduplication logic without network calls.
CompletableFuture<ClientCnx> neverCompletes = new CompletableFuture<>();
when(cnxPool.getConnection(any(InetSocketAddress.class))).thenReturn(neverCompletes);
@@ -291,7 +310,7 @@ public class BinaryProtoLookupServiceTest {
when(client.instrumentProvider()).thenReturn(InstrumentProvider.NOOP);
when(client.getCnxPool()).thenReturn(cnxPool);
- // Never-completing connection prevents the thenAcceptAsync callback in getTopicsUnderNamespace from executing,
+ // Never-completing connection prevents the thenComposeAsync callback in getTopicsUnderNamespace from executing,
// isolating only the deduplication logic without network calls.
CompletableFuture<ClientCnx> neverCompletes = new CompletableFuture<>();
when(cnxPool.getConnection(any(InetSocketAddress.class))).thenReturn(neverCompletes);
@@ -362,7 +381,7 @@ public class BinaryProtoLookupServiceTest {
when(client.instrumentProvider()).thenReturn(InstrumentProvider.NOOP);
when(client.getCnxPool()).thenReturn(cnxPool);
- // Never-completing connection prevents the thenAcceptAsync callback in getPartitionedTopicMetadata
+ // Never-completing connection prevents the thenComposeAsync callback in getPartitionedTopicMetadata
// from executing, isolating only the deduplication logic without network calls.
CompletableFuture<ClientCnx> neverCompletes = new CompletableFuture<>();
when(cnxPool.getConnection(any(InetSocketAddress.class))).thenReturn(neverCompletes);
@@ -422,7 +441,7 @@ public class BinaryProtoLookupServiceTest {
when(client.instrumentProvider()).thenReturn(InstrumentProvider.NOOP);
when(client.getCnxPool()).thenReturn(cnxPool);
- // Never-completing connection prevents the thenAcceptAsync callback in getPartitionedTopicMetadata
+ // Never-completing connection prevents the thenComposeAsync callback in getPartitionedTopicMetadata
// from executing, so the test exercises only the deduplication map without network calls
CompletableFuture<ClientCnx> neverCompletes = new CompletableFuture<>();
when(cnxPool.getConnection(any(InetSocketAddress.class))).thenReturn(neverCompletes);
@@ -473,4 +492,147 @@ public class BinaryProtoLookupServiceTest {
scheduler.shutdownNow();
}
}
+
+ @Test
+ public void testGetTopicsRetriesAfterConnectionClosedMidRequest() throws Exception {
+ GetTopicsFixture fixture = new GetTopicsFixture(30000);
+ CompletableFuture<GetTopicsResult> getTopicsFuture = fixture.getTopics();
+
+ CommandGetTopicsOfNamespace firstRequest = fixture.readGetTopicsRequest();
+ assertThat(fixture.scheduledTasks).isEmpty();
+
+ // the connection drops before the broker answered
+ fixture.channel.close().get();
+ assertThat(getTopicsFuture).isNotDone();
+ assertThat(fixture.scheduledTasks).hasSize(1);
+
+ // the retry sends the request again on a new connection
+ fixture.runScheduledTasks();
+ CommandGetTopicsOfNamespace secondRequest = fixture.readGetTopicsRequest();
+ assertThat(secondRequest.getRequestId()).isNotEqualTo(firstRequest.getRequestId());
+
+ List<String> topics = Collections.singletonList("persistent://public/default/test");
+ fixture.channel.writeInbound(Commands.serializeWithSize(Commands.newGetTopicsOfNamespaceResponseCommand(
+ topics, TopicList.calculateHash(topics), true, true, secondRequest.getRequestId())));
+
+ assertThat(getTopicsFuture).succeedsWithin(Duration.ofSeconds(1))
+ .satisfies(result -> assertThat(result.getTopics()).containsExactlyElementsOf(topics));
+ }
+
+ @Test
+ public void testGetTopicsFailsWithTimeoutWhenNoTimeIsLeftToRetry() throws Exception {
+ // a zero operation timeout leaves no time for a retry after the first failure
+ GetTopicsFixture fixture = new GetTopicsFixture(0);
+ CompletableFuture<GetTopicsResult> getTopicsFuture = fixture.getTopics();
+ fixture.readGetTopicsRequest();
+
+ fixture.channel.close().get();
+
+ assertThat(fixture.scheduledTasks).isEmpty();
+ assertThat(getTopicsFuture).failsWithin(Duration.ofSeconds(1))
+ .withThrowableOfType(ExecutionException.class)
+ .havingCause()
+ .isInstanceOf(PulsarClientException.TimeoutException.class)
+ .havingCause()
+ .isInstanceOf(PulsarClientException.ConnectException.class);
+ }
+
+ @Test
+ public void testGetTopicsDoesNotRetryNonRetriableError() throws Exception {
+ GetTopicsFixture fixture = new GetTopicsFixture(30000);
+ CompletableFuture<GetTopicsResult> getTopicsFuture = fixture.getTopics();
+ CommandGetTopicsOfNamespace request = fixture.readGetTopicsRequest();
+
+ fixture.channel.writeInbound(
+ Commands.newError(request.getRequestId(), ServerError.AuthorizationError, "not authorized"));
+
+ assertThat(fixture.scheduledTasks).isEmpty();
+ assertThat(getTopicsFuture).failsWithin(Duration.ofSeconds(1))
+ .withThrowableOfType(ExecutionException.class)
+ .havingCause()
+ .isInstanceOf(PulsarClientException.AuthorizationException.class);
+ }
+
+ /**
+ * Runs getTopicsUnderNamespace against a real {@link ClientCnx} on an {@link EmbeddedChannel}. Lookup callbacks
+ * run inline and retries are captured instead of scheduled, so a test drives every step itself.
+ */
+ private static class GetTopicsFixture {
+ private final List<Runnable> scheduledTasks = new ArrayList<>();
+ private final BinaryProtoLookupService lookupService;
+ private EmbeddedChannel channel;
+
+ GetTopicsFixture(long operationTimeoutMs) throws Exception {
+ ClientConfigurationData conf = new ClientConfigurationData();
+ conf.setOperationTimeoutMs(operationTimeoutMs);
+
+ EventLoopGroup eventLoopGroup = mock(EventLoopGroup.class);
+ doReturn(mock(io.netty.util.concurrent.ScheduledFuture.class)).when(eventLoopGroup)
+ .scheduleAtFixedRate(any(Runnable.class), anyLong(), anyLong(), any(TimeUnit.class));
+
+ ConnectionPool cnxPool = mock(ConnectionPool.class);
+ doAnswer(invocation -> {
+ ClientCnx cnx = new ClientCnx(InstrumentProvider.NOOP, conf, eventLoopGroup);
+ channel = new EmbeddedChannel(new ChannelInitializer<Channel>() {
+ @Override
+ protected void initChannel(Channel ch) {
+ FrameDecoderUtil.addFrameDecoder(ch.pipeline(), Commands.DEFAULT_MAX_MESSAGE_SIZE);
+ ch.pipeline().addLast(cnx);
+ }
+ });
+ assertThat(readCommand().getType()).isEqualTo(Type.CONNECT);
+ channel.writeInbound(Commands.newConnected(Commands.getCurrentProtocolVersion(), true));
+ return CompletableFuture.completedFuture(cnx);
+ }).when(cnxPool).getConnection(any(InetSocketAddress.class));
+
+ AtomicLong requestIds = new AtomicLong();
+ PulsarClientImpl client = mock(PulsarClientImpl.class);
+ doReturn(InstrumentProvider.NOOP).when(client).instrumentProvider();
+ doReturn(cnxPool).when(client).getCnxPool();
+ doReturn(conf).when(client).getConfiguration();
+ doAnswer(invocation -> requestIds.incrementAndGet()).when(client).newRequestId();
+
+ ScheduledExecutorService scheduler = mock(ScheduledExecutorService.class);
+ doAnswer(invocation -> {
+ scheduledTasks.add(invocation.getArgument(0));
+ return mock(ScheduledFuture.class);
+ }).when(scheduler).schedule(any(Runnable.class), anyLong(), any(TimeUnit.class));
+
+ lookupService = new BinaryProtoLookupService(client, "pulsar://localhost:6650", null, false, scheduler,
+ MoreExecutors.newDirectExecutorService());
+ }
+
+ CompletableFuture<GetTopicsResult> getTopics() {
+ return lookupService.getTopicsUnderNamespace(NamespaceName.get("public", "default"), Mode.PERSISTENT,
+ ".*", null);
+ }
+
+ void runScheduledTasks() {
+ List<Runnable> tasks = new ArrayList<>(scheduledTasks);
+ scheduledTasks.clear();
+ tasks.forEach(Runnable::run);
+ }
+
+ CommandGetTopicsOfNamespace readGetTopicsRequest() {
+ BaseCommand cmd = readCommand();
+ assertThat(cmd.getType()).isEqualTo(Type.GET_TOPICS_OF_NAMESPACE);
+ return cmd.getGetTopicsOfNamespace();
+ }
+
+ private BaseCommand readCommand() {
+ ByteBuf frame = channel.readOutbound();
+ assertNotNull(frame);
+ try {
+ frame.skipBytes(4); // total frame size
+ int cmdSize = (int) frame.readUnsignedInt();
+ BaseCommand cmd = new BaseCommand();
+ // Parse a copy: the command's fields are decoded lazily from the buffer, which is released below
+ cmd.parseFrom(Unpooled.wrappedBuffer(ByteBufUtil.getBytes(frame, frame.readerIndex(), cmdSize)),
+ cmdSize);
+ return cmd;
+ } finally {
+ frame.release();
+ }
+ }
+ }
}