Skip to content

Commit bde1410

Browse files
lhotarilhotari
committed
Fix get topics of namespace retries
1 parent 2b56eba commit bde1410

2 files changed

Lines changed: 177 additions & 45 deletions

File tree

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

Lines changed: 40 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,6 @@
3434
import java.util.concurrent.Executors;
3535
import java.util.concurrent.ScheduledExecutorService;
3636
import java.util.concurrent.TimeUnit;
37-
import java.util.concurrent.atomic.AtomicLong;
3837
import org.apache.commons.lang3.mutable.MutableObject;
3938
import org.apache.commons.lang3.tuple.Pair;
4039
import org.apache.pulsar.client.api.PulsarClientException;
@@ -404,64 +403,66 @@ public CompletableFuture<GetTopicsResult> getTopicsUnderNamespace(NamespaceName
404403
String topicsHash) {
405404
CompletableFuture<GetTopicsResult> topicsFuture = new CompletableFuture<>();
406405

407-
AtomicLong opTimeoutMs = new AtomicLong(client.getConfiguration().getOperationTimeoutMs());
406+
long opTimeoutMs = client.getConfiguration().getOperationTimeoutMs();
408407
Backoff backoff = new BackoffBuilder()
409408
.setInitialTime(100, TimeUnit.MILLISECONDS)
410-
.setMandatoryStop(opTimeoutMs.get() * 2, TimeUnit.MILLISECONDS)
409+
.setMandatoryStop(opTimeoutMs * 2, TimeUnit.MILLISECONDS)
411410
.setMax(1, TimeUnit.MINUTES)
412411
.create();
413-
getTopicsUnderNamespace(namespace, backoff, opTimeoutMs, topicsFuture, mode,
412+
long startTimeNanos = System.nanoTime();
413+
long retryUntilNanos = startTimeNanos + TimeUnit.MILLISECONDS.toNanos(opTimeoutMs);
414+
getTopicsUnderNamespace(namespace, backoff, startTimeNanos, retryUntilNanos, topicsFuture, mode,
414415
topicsPattern, topicsHash);
415416
return topicsFuture;
416417
}
417418

418419
private void getTopicsUnderNamespace(
419420
NamespaceName namespace,
420421
Backoff backoff,
421-
AtomicLong remainingTime,
422+
long startTimeNanos,
423+
long retryUntilNanos,
422424
CompletableFuture<GetTopicsResult> getTopicsResultFuture,
423425
Mode mode,
424426
String topicsPattern,
425427
String topicsHash) {
426-
long startTime = System.nanoTime();
427-
428-
client.getCnxPool().getConnection(serviceNameResolver).thenAcceptAsync(clientCnx -> {
428+
client.getCnxPool().getConnection(serviceNameResolver).thenComposeAsync(clientCnx -> {
429429
long requestId = client.newRequestId();
430-
ByteBuf request = Commands.newGetTopicsOfNamespaceRequest(
431-
namespace.toString(), requestId, mode, topicsPattern, topicsHash);
432-
433-
clientCnx.newGetTopicsOfNamespace(request, requestId).whenComplete((r, t) -> {
434-
if (t != null) {
435-
histoListTopics.recordFailure(System.nanoTime() - startTime);
436-
getTopicsResultFuture.completeExceptionally(t);
430+
ByteBuf request = Commands.newGetTopicsOfNamespaceRequest(namespace.toString(), requestId, mode,
431+
topicsPattern, topicsHash);
432+
return clientCnx.newGetTopicsOfNamespace(request, requestId).whenComplete((r, t) -> {
433+
client.getCnxPool().releaseConnection(clientCnx);
434+
});
435+
}, lookupPinnedExecutor).whenComplete((r, t) -> {
436+
if (t != null) {
437+
long nowNanos = System.nanoTime();
438+
if (nowNanos > retryUntilNanos) {
439+
histoListTopics.recordFailure(System.nanoTime() - startTimeNanos);
440+
log.warn("[namespace: {}] Error while getTopicsUnderNamespace -- No more retries left."
441+
+ " Last error was {}", namespace, t.getMessage());
442+
getTopicsResultFuture.completeExceptionally(new PulsarClientException.TimeoutException(
443+
format("Could not get topics of namespace %s within configured timeout",
444+
namespace.toString())));
437445
} else {
438-
histoListTopics.recordSuccess(System.nanoTime() - startTime);
439-
if (log.isDebugEnabled()) {
440-
log.debug("[namespace: {}] Success get topics list in request: {}",
441-
namespace, requestId);
446+
long nextDelay = backoff.next();
447+
log.warn("[namespace: {}] Error while getTopicsUnderNamespace -- Will try again in"
448+
+ " {} ms. Error was {}", namespace, nextDelay, t.getMessage());
449+
if (!getTopicsResultFuture.isDone()) {
450+
scheduleExecutor.schedule(() -> {
451+
getTopicsUnderNamespace(namespace, backoff, startTimeNanos, retryUntilNanos,
452+
getTopicsResultFuture, mode, topicsPattern, topicsHash);
453+
}, nextDelay, TimeUnit.MILLISECONDS);
454+
} else {
455+
log.info("[namespace: {}] Ignoring retry in getTopicsUnderNamespace -- Future is already "
456+
+ "completed", namespace);
442457
}
443-
getTopicsResultFuture.complete(r);
444458
}
445-
client.getCnxPool().releaseConnection(clientCnx);
446-
});
447-
}, lookupPinnedExecutor).exceptionally((e) -> {
448-
long nextDelay = Math.min(backoff.next(), remainingTime.get());
449-
if (nextDelay <= 0) {
450-
getTopicsResultFuture.completeExceptionally(
451-
new PulsarClientException.TimeoutException(
452-
format("Could not get topics of namespace %s within configured timeout",
453-
namespace.toString())));
454-
return null;
459+
} else {
460+
histoListTopics.recordSuccess(System.nanoTime() - startTimeNanos);
461+
if (log.isDebugEnabled()) {
462+
log.debug("[namespace: {}] Success get topics list", namespace);
463+
}
464+
getTopicsResultFuture.complete(r);
455465
}
456-
457-
scheduleExecutor.schedule(() -> {
458-
log.warn("[namespace: {}] Could not get connection while getTopicsUnderNamespace -- Will try again in"
459-
+ " {} ms", namespace, nextDelay);
460-
remainingTime.addAndGet(-nextDelay);
461-
getTopicsUnderNamespace(namespace, backoff, remainingTime, getTopicsResultFuture,
462-
mode, topicsPattern, topicsHash);
463-
}, nextDelay, TimeUnit.MILLISECONDS);
464-
return null;
465466
});
466467
}
467468

‎pulsar-client/src/test/java/org/apache/pulsar/client/impl/BinaryProtoLookupServiceTest.java‎

Lines changed: 137 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
*/
1919
package org.apache.pulsar.client.impl;
2020

21+
import static org.assertj.core.api.Assertions.assertThat;
2122
import static org.mockito.Mockito.any;
2223
import static org.mockito.Mockito.anyLong;
2324
import static org.mockito.Mockito.doAnswer;
@@ -35,25 +36,38 @@
3536
import static org.testng.Assert.fail;
3637
import com.google.common.util.concurrent.MoreExecutors;
3738
import io.netty.buffer.ByteBuf;
39+
import io.netty.channel.ChannelHandler;
40+
import io.netty.channel.embedded.EmbeddedChannel;
3841
import io.netty.util.concurrent.DefaultThreadFactory;
42+
import io.netty.util.concurrent.ScheduledFuture;
3943
import java.lang.reflect.Field;
4044
import java.net.InetSocketAddress;
45+
import java.time.Duration;
46+
import java.util.ArrayList;
47+
import java.util.Collections;
48+
import java.util.List;
4149
import java.util.concurrent.CompletableFuture;
4250
import java.util.concurrent.ExecutionException;
4351
import java.util.concurrent.ExecutorService;
4452
import java.util.concurrent.Executors;
4553
import java.util.concurrent.ScheduledExecutorService;
4654
import java.util.concurrent.atomic.AtomicInteger;
4755
import java.util.concurrent.atomic.AtomicReference;
56+
import java.util.function.Function;
4857
import org.apache.pulsar.client.api.PulsarClientException.LookupException;
4958
import org.apache.pulsar.client.impl.BinaryProtoLookupService.LookupDataResult;
5059
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
5160
import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
5261
import org.apache.pulsar.common.api.proto.BaseCommand;
5362
import org.apache.pulsar.common.api.proto.BaseCommand.Type;
63+
import org.apache.pulsar.common.api.proto.CommandConnect;
64+
import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace;
65+
import org.apache.pulsar.common.lookup.GetTopicsResult;
66+
import org.apache.pulsar.common.naming.NamespaceName;
5467
import org.apache.pulsar.common.naming.TopicName;
5568
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
5669
import org.apache.pulsar.common.protocol.Commands;
70+
import org.apache.pulsar.common.topics.TopicList;
5771
import org.awaitility.Awaitility;
5872
import org.testng.annotations.AfterMethod;
5973
import org.testng.annotations.BeforeMethod;
@@ -64,6 +78,8 @@ public class BinaryProtoLookupServiceTest {
6478
private TopicName topicName;
6579
private ExecutorService internalExecutor;
6680
private ScheduledExecutorService scheduledExecutorService;
81+
private ConnectionPool cnxPool;
82+
private ClientConfigurationData conf;
6783

6884
@AfterMethod
6985
public void cleanup() throws Exception {
@@ -94,25 +110,26 @@ public void setup() throws Exception {
94110

95111
CompletableFuture<ClientCnx> connectionFuture = CompletableFuture.completedFuture(clientCnx);
96112

97-
ConnectionPool cnxPool = mock(ConnectionPool.class);
113+
cnxPool = mock(ConnectionPool.class);
98114
when(cnxPool.getConnection(any(InetSocketAddress.class))).thenReturn(connectionFuture);
99115
when(cnxPool.getConnection(any(ServiceNameResolver.class))).thenReturn(connectionFuture);
100116

101-
ClientConfigurationData clientConfig = mock(ClientConfigurationData.class);
102-
doReturn(0).when(clientConfig).getMaxLookupRedirects();
117+
conf = new ClientConfigurationData();
118+
conf.setMaxLookupRedirects(0);
103119

104120
PulsarClientImpl client = mock(PulsarClientImpl.class);
105121
doReturn(InstrumentProvider.NOOP).when(client).instrumentProvider();
106122
doReturn(cnxPool).when(client).getCnxPool();
107-
doReturn(clientConfig).when(client).getConfiguration();
123+
doReturn(conf).when(client).getConfiguration();
108124
doReturn(1L).when(client).newRequestId();
109-
ClientConfigurationData data = new ClientConfigurationData();
110-
doReturn(data).when(client).getConfiguration();
111125
internalExecutor =
112126
Executors.newSingleThreadExecutor(new DefaultThreadFactory("pulsar-client-test-internal-executor"));
113127
doReturn(internalExecutor).when(client).getInternalExecutorService();
114128

115129
scheduledExecutorService = mock(ScheduledExecutorService.class);
130+
// just return a mock ScheduledFuture for scheduleAtFixedRate by default without any action
131+
doReturn(mock(ScheduledFuture.class)).when(scheduledExecutorService)
132+
.scheduleAtFixedRate(any(), anyLong(), anyLong(), any());
116133

117134
ExecutorService lookupExecutor = MoreExecutors.newDirectExecutorService();
118135
lookup = spy(new BinaryProtoLookupService(client, "pulsar://localhost:6650", null,
@@ -341,4 +358,118 @@ public void testPartitionedMetadataDeduplicationDifferentParameterCombinations()
341358
scheduler.shutdownNow();
342359
}
343360
}
361+
362+
@Test
363+
public void testGetTopicsRetries() throws ExecutionException, InterruptedException {
364+
AtomicReference<EmbeddedChannel> channelRef = configureEmbeddedChannel();
365+
366+
List<Runnable> scheduledTasks = configureScheduledTasks();
367+
368+
NamespaceName ns = NamespaceName.get("public", "default");
369+
CommandGetTopicsOfNamespace.Mode mode = CommandGetTopicsOfNamespace.Mode.PERSISTENT;
370+
String pattern = ".*";
371+
String topicsHash = null;
372+
CompletableFuture<GetTopicsResult> getTopicsFuture =
373+
lookup.getTopicsUnderNamespace(ns, mode, pattern, topicsHash);
374+
375+
CommandGetTopicsOfNamespace cmdGetTopicsOfNamespace =
376+
readCommand(channelRef.get(), Type.GET_TOPICS_OF_NAMESPACE, BaseCommand::getGetTopicsOfNamespace);
377+
assertNotNull(cmdGetTopicsOfNamespace);
378+
379+
// no tasks
380+
assertThat(scheduledTasks).isEmpty();
381+
382+
// close embedded connect
383+
channelRef.get().close().get();
384+
385+
// the future shouldn't be completed yet
386+
assertThat(getTopicsFuture).isNotDone();
387+
388+
// the task to retry has been scheduled
389+
assertThat(scheduledTasks).hasSize(1);
390+
391+
// run tasks
392+
scheduledTasks.forEach(Runnable::run);
393+
394+
List<String> topics = Collections.singletonList("persistent://public/default/test");
395+
String topicsHash2 = TopicList.calculateHash(topics);
396+
channelRef.get().writeInbound(Commands.serializeWithSize(
397+
Commands.newGetTopicsOfNamespaceResponseCommand(topics, topicsHash2, true, true,
398+
cmdGetTopicsOfNamespace.getRequestId())));
399+
400+
assertThat(getTopicsFuture).succeedsWithin(Duration.ofSeconds(1))
401+
.satisfies(topicsResult -> {
402+
assertThat(topicsResult.getTopics()).containsExactly("persistent://public/default/test");
403+
});
404+
}
405+
406+
@Test
407+
public void testGetTopicsRetriesExpire() throws ExecutionException, InterruptedException {
408+
// set minimal timeout so that retries would expire
409+
conf.setOperationTimeoutMs(1);
410+
411+
AtomicReference<EmbeddedChannel> channelRef = configureEmbeddedChannel();
412+
413+
List<Runnable> scheduledTasks = configureScheduledTasks();
414+
415+
NamespaceName ns = NamespaceName.get("public", "default");
416+
CommandGetTopicsOfNamespace.Mode mode = CommandGetTopicsOfNamespace.Mode.PERSISTENT;
417+
String pattern = ".*";
418+
String topicsHash = null;
419+
CompletableFuture<GetTopicsResult> getTopicsFuture =
420+
lookup.getTopicsUnderNamespace(ns, mode, pattern, topicsHash);
421+
422+
CommandGetTopicsOfNamespace cmdGetTopicsOfNamespace =
423+
readCommand(channelRef.get(), Type.GET_TOPICS_OF_NAMESPACE, BaseCommand::getGetTopicsOfNamespace);
424+
assertNotNull(cmdGetTopicsOfNamespace);
425+
426+
// no tasks
427+
assertThat(scheduledTasks).isEmpty();
428+
429+
// operation would have already expired and won't retry
430+
Thread.sleep(100);
431+
432+
// close embedded connect
433+
channelRef.get().close().get();
434+
435+
// the future should have been completed exceptionally
436+
assertThat(getTopicsFuture).isCompletedExceptionally();
437+
}
438+
439+
private List<Runnable> configureScheduledTasks() {
440+
List<Runnable> tasks = Collections.synchronizedList(new ArrayList<>());
441+
doAnswer(invocationOnMock -> {
442+
Runnable runnable = invocationOnMock.getArgument(0);
443+
tasks.add(runnable);
444+
return mock(ScheduledFuture.class);
445+
}).when(scheduledExecutorService)
446+
.schedule(any(Runnable.class), anyLong(), any());
447+
return tasks;
448+
}
449+
450+
private AtomicReference<EmbeddedChannel> configureEmbeddedChannel() {
451+
AtomicReference<EmbeddedChannel> channelRef = new AtomicReference<>();
452+
453+
doAnswer(invocationOnMock -> {
454+
ClientCnx clientCnx = new ClientCnx(InstrumentProvider.NOOP, conf, scheduledExecutorService);
455+
ChannelHandler clientCnxChannelHandler = clientCnx;
456+
EmbeddedChannel channel = new EmbeddedChannel(clientCnxChannelHandler);
457+
channelRef.set(channel);
458+
CommandConnect cmd = readCommand(channel, Type.CONNECT, BaseCommand::getConnect);
459+
assertNotNull(cmd);
460+
channel.writeInbound(Commands.newConnected(Commands.getCurrentProtocolVersion(), true));
461+
return CompletableFuture.completedFuture(clientCnx);
462+
}).when(cnxPool).getConnection(any(ServiceNameResolver.class));
463+
return channelRef;
464+
}
465+
466+
private <T> T readCommand(EmbeddedChannel channel, BaseCommand.Type expectedType,
467+
Function<BaseCommand, T> extractFunction) {
468+
ByteBuf buffer = (ByteBuf) channel.outboundMessages().remove();
469+
BaseCommand cmd = new BaseCommand();
470+
int cmdSize = (int) buffer.readUnsignedInt();
471+
cmd.parseFrom(buffer, cmdSize);
472+
assertThat(cmd.getType()).isEqualTo(expectedType);
473+
return extractFunction.apply(cmd);
474+
}
344475
}

0 commit comments

Comments
 (0)