Skip to content

[improve][client] Make seek wait for reconnect #18202

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
nodece wants to merge 1 commit into from
Closed

[improve][client] Make seek wait for reconnect #18202

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 52 additions & 25 deletions pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,6 @@
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLong;
Expand Down Expand Up @@ -156,7 +155,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
private volatile BatchMessageIdImpl startMessageId;

private volatile BatchMessageIdImpl seekMessageId;
private final AtomicBoolean duringSeek;
private CompletableFuture<Void> seekFuture = null;

private final BatchMessageIdImpl initialStartMessageId;

Expand Down Expand Up @@ -290,8 +289,6 @@ protected ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurat
stats = ConsumerStatsDisabled.INSTANCE;
}

duringSeek = new AtomicBoolean(false);

if (conf.getAckTimeoutMillis() != 0) {
if (conf.getAckTimeoutRedeliveryBackoff() != null) {
this.unAckedMessageTracker = new UnAckedMessageRedeliveryTracker(client, this, conf);
Expand Down Expand Up @@ -787,9 +784,6 @@ public void connectionOpened(final ClientCnx cnx) {
topic, subscription, cnx.ctx().channel(), consumerId);

long requestId = client.newRequestId();
if (duringSeek.get()) {
acknowledgmentsGroupingTracker.flushAndClean();
}

SUBSCRIBE_DEADLINE_UPDATER
.compareAndSet(this, 0L, System.currentTimeMillis()
Expand Down Expand Up @@ -839,6 +833,11 @@ public void connectionOpened(final ClientCnx cnx) {
// Use the current epoch to subscribe.
conf.getSubscriptionProperties(), CONSUMER_EPOCH.get(this));

boolean duringSeek = isDuringSeek();
if (duringSeek) {
seekFuture.complete(null);
}

cnx.sendRequestWithId(request, requestId).thenRun(() -> {
synchronized (ConsumerImpl.this) {
if (changeToReadyState()) {
Expand All @@ -854,6 +853,8 @@ public void connectionOpened(final ClientCnx cnx) {
}
}

seekMessageId = null;

resetBackoff();

boolean firstTimeConnect = subscribeFuture.complete(this);
Expand Down Expand Up @@ -932,7 +933,7 @@ private BatchMessageIdImpl clearReceiverQueue() {
incomingMessages.drainTo(currentMessageQueue);
resetIncomingMessageSize();

if (duringSeek.compareAndSet(true, false)) {
if (seekMessageId != null) {
return seekMessageId;
} else if (subscriptionMode == SubscriptionMode.Durable) {
return startMessageId;
Expand Down Expand Up @@ -2142,36 +2143,62 @@ private Optional<CompletableFuture<Void>> seekAsyncCheckState(String seekBy) {
return Optional.empty();
}

private CompletableFuture<Void> seekAsyncInternal(long requestId, ByteBuf seek, MessageId seekId, String seekBy) {
final CompletableFuture<Void> seekFuture = new CompletableFuture<>();
private void resetSeekFuture(CompletableFuture<Void> doneFuture) {
seekFuture = new CompletableFuture<>();
seekFuture.whenComplete((__, ex) -> {
if (ex != null) {
doneFuture.completeExceptionally(ex);
return;
}
acknowledgmentsGroupingTracker.flushAndClean();
lastDequeuedMessageId = MessageId.earliest;
clearIncomingMessages();
doneFuture.complete(null);
log.info("[{}][{}] Successfully reset subscription", topic, subscription);
});
}

private boolean isDuringSeek() {
return seekFuture != null && !seekFuture.isDone();
}

private CompletableFuture<Void> seekAsyncInternal(long requestId, ByteBuf seek, MessageId seekId,
String seekBy) {
CompletableFuture<Void> future = new CompletableFuture<>();

boolean duringSeek = true;
if (seekFuture == null || seekFuture.isDone()) {
resetSeekFuture(future);
duringSeek = false;
}

if (duringSeek) {
seekFuture.join();
resetSeekFuture(future);
}

ClientCnx cnx = cnx();

BatchMessageIdImpl originSeekMessageId = seekMessageId;
seekMessageId = new BatchMessageIdImpl((MessageIdImpl) seekId);
duringSeek.set(true);
log.info("[{}][{}] Seeking subscription to {}", topic, subscription, seekBy);

cnx.sendRequestWithId(seek, requestId).thenRun(() -> {
log.info("[{}][{}] Successfully reset subscription to {}", topic, subscription, seekBy);
acknowledgmentsGroupingTracker.flushAndClean();

lastDequeuedMessageId = MessageId.earliest;

clearIncomingMessages();
seekFuture.complete(null);
log.info("[{}][{}] Successfully sent the seek command to reset subscription to {}", topic, subscription,
seekBy);
}).exceptionally(e -> {
// re-set duringSeek and seekMessageId if seek failed
// re-set seekMessageId if seek failed
seekMessageId = originSeekMessageId;
duringSeek.set(false);
log.error("[{}][{}] Failed to reset subscription: {}", topic, subscription, e.getCause().getMessage());
log.error("[{}][{}] Failed to send the seek command to reset subscription: {}", topic, subscription,
e.getCause().getMessage());

seekFuture.completeExceptionally(
PulsarClientException.wrap(e.getCause(),
seekFuture.completeExceptionally(PulsarClientException.wrap(e.getCause(),
String.format("Failed to seek the subscription %s of the topic %s to %s",
subscription, topicName.toString(), seekBy)));
subscription, topicName.toString(), seekBy)));
return null;
});
return seekFuture;

return future;
}

@Override
Expand Down