Skip to content

Fixed reader cursor seek with disconnection #5278

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
merlimat wants to merge 3 commits into from
Closed

Fixed reader cursor seek with disconnection #5278

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
267ab18
Fixed reader cursor seek with disconnection
merlimat Aug 27, 2019
742fd2e
Merge remote-tracking branch 'apache/master' into seek-without-close
merlimat Oct 31, 2019
6da6585
Merge branch 'master' into seek-without-close
tuteng Jan 17, 2020
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
6 changes: 4 additions & 2 deletions managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -899,7 +899,8 @@ public void operationComplete() {
log.info("[{}] reset position to {} skipping from current read position {} on cursor {}",
ledger.getName(), newPosition, oldReadPosition, name);
}
readPosition = newPosition;

rewind();
} finally {
lock.writeLock().unlock();
}
Expand All @@ -910,8 +911,9 @@ public void operationComplete() {
ledger.getName(), newPosition, name);
}
}
callback.resetComplete(newPosition);

callback.resetComplete(newPosition);
notifyEntriesAvailable();
}

@Override
Expand Down
93 changes: 33 additions & 60 deletions ...ker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -683,70 +683,43 @@ private void resetCursor(Position finalPosition, CompletableFuture<Void> future)
return;
}

final CompletableFuture<Void> disconnectFuture;

// Lock the Subscription object before locking the Dispatcher object to avoid deadlocks
synchronized (this) {
if (dispatcher != null && dispatcher.isConsumerConnected()) {
disconnectFuture = dispatcher.disconnectAllConsumers(true);
} else {
disconnectFuture = CompletableFuture.completedFuture(null);
}
}

disconnectFuture.whenComplete((aVoid, throwable) -> {
if (dispatcher != null) {
dispatcher.resetCloseFuture();
}

if (throwable != null) {
log.error("[{}][{}] Failed to disconnect consumer from subscription", topicName, subName, throwable);
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
future.completeExceptionally(
new SubscriptionBusyException("Failed to disconnect consumers from subscription"));
return;
}

log.info("[{}][{}] Successfully disconnected consumers from subscription, proceeding with cursor reset",
topicName, subName);
try {
cursor.asyncResetCursor(finalPosition, new AsyncCallbacks.ResetCursorCallback() {
@Override
public void resetComplete(Object ctx) {
if (log.isDebugEnabled()) {
log.debug("[{}][{}] Successfully reset subscription to position {}", topicName, subName,
finalPosition);
}

try {
cursor.asyncResetCursor(finalPosition, new AsyncCallbacks.ResetCursorCallback() {
@Override
public void resetComplete(Object ctx) {
if (log.isDebugEnabled()) {
log.debug("[{}][{}] Successfully reset subscription to position {}", topicName, subName,
finalPosition);
}
if (dispatcher != null) {
dispatcher.cursorIsReset();
}
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
future.complete(null);
if (dispatcher != null) {
dispatcher.cursorIsReset();
}
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
future.complete(null);
}

@Override
public void resetFailed(ManagedLedgerException exception, Object ctx) {
log.error("[{}][{}] Failed to reset subscription to position {}", topicName, subName,
finalPosition, exception);
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
// todo - retry on InvalidCursorPositionException
// or should we just ask user to retry one more time?
if (exception instanceof InvalidCursorPositionException) {
future.completeExceptionally(new SubscriptionInvalidCursorPosition(exception.getMessage()));
} else if (exception instanceof ConcurrentFindCursorPositionException) {
future.completeExceptionally(new SubscriptionBusyException(exception.getMessage()));
} else {
future.completeExceptionally(new BrokerServiceException(exception));
}
@Override
public void resetFailed(ManagedLedgerException exception, Object ctx) {
log.error("[{}][{}] Failed to reset subscription to position {}", topicName, subName,
finalPosition, exception);
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
// todo - retry on InvalidCursorPositionException
// or should we just ask user to retry one more time?
if (exception instanceof InvalidCursorPositionException) {
future.completeExceptionally(new SubscriptionInvalidCursorPosition(exception.getMessage()));
} else if (exception instanceof ConcurrentFindCursorPositionException) {
future.completeExceptionally(new SubscriptionBusyException(exception.getMessage()));
} else {
future.completeExceptionally(new BrokerServiceException(exception));
}
});
} catch (Exception e) {
log.error("[{}][{}] Error while resetting cursor", topicName, subName, e);
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
future.completeExceptionally(new BrokerServiceException(e));
}
});
}
});
} catch (Exception e) {
log.error("[{}][{}] Error while resetting cursor", topicName, subName, e);
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
future.completeExceptionally(new BrokerServiceException(e));
}
}

@Override
Expand Down
1 change: 0 additions & 1 deletion pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1748,7 +1748,6 @@ public void persistentTopicsInvalidCursorReset() throws Exception {
}

admin.topics().resetCursor(topicName, "my-sub", System.currentTimeMillis() + 90000);
consumer = client.newConsumer().topic(topicName).subscriptionName("my-sub").subscribe();
consumer.close();
client.close();

Expand Down
1 change: 0 additions & 1 deletion pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1665,7 +1665,6 @@ public void persistentTopicsInvalidCursorReset() throws Exception {
}

admin.topics().resetCursor(topicName, "my-sub", System.currentTimeMillis() + 90000);
consumer = client.newConsumer().topic(topicName).subscriptionName("my-sub").subscribe();
consumer.close();
client.close();

Expand Down
11 changes: 8 additions & 3 deletions pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,17 +25,22 @@

import com.google.common.collect.Lists;
import com.google.common.collect.Sets;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

import org.apache.pulsar.client.impl.BatchMessageIdImpl;
import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.client.impl.ReaderImpl;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.RelativeTimeUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -638,7 +643,7 @@ public void testReaderIsAbleToSeekWithMessageIdOnMiddleOfTopic() throws Exceptio

// Read all halved messages after seek()
Set<String> messageSetB = Sets.newHashSet();
for (int i = halfMessages + 1; i < numOfMessage; i++) {
for (int i = halfMessages; i < numOfMessage; i++) {
Message<byte[]> message = reader.readNext();
String receivedMessage = new String(message.getData());
String expectedMessage = String.format("msg num %d", i);
Expand Down