Skip to content

[pulsar-broker] cursor: safe guard to avoid cursor-znode after cursor is closed #3556

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

Merged
sijie merged 1 commit into from
Feb 9, 2019
Merged

[pulsar-broker] cursor: safe guard to avoid cursor-znode after cursor is closed #3556

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
30 changes: 22 additions & 8 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 @@ -166,6 +166,7 @@ enum State {
NoLedger, // There is no metadata ledger open for writing
Open, // Metadata ledger is ready
SwitchingLedger, // The metadata ledger is being switched
Closing, // The managed cursor is closing
Closed // The managed cursor has been closed
}

Expand Down Expand Up @@ -439,7 +440,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
public void asyncReadEntries(final int numberOfEntriesToRead, final ReadEntriesCallback callback,
final Object ctx) {
checkArgument(numberOfEntriesToRead > 0);
if (STATE_UPDATER.get(this) == State.Closed) {
if (isClosed()) {
callback.readEntriesFailed(new ManagedLedgerException("Cursor was already closed"), ctx);
return;
}
Expand Down Expand Up @@ -489,7 +490,7 @@ public void readEntryComplete(Entry entry, Object ctx) {
public void asyncGetNthEntry(int n, IndividualDeletedEntries deletedEntries, ReadEntryCallback callback,
Object ctx) {
checkArgument(n > 0);
if (STATE_UPDATER.get(this) == State.Closed) {
if (isClosed()) {
callback.readEntryFailed(new ManagedLedgerException("Cursor was already closed"), ctx);
return;
}
Expand Down Expand Up @@ -554,7 +555,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
@Override
public void asyncReadEntriesOrWait(int numberOfEntriesToRead, ReadEntriesCallback callback, Object ctx) {
checkArgument(numberOfEntriesToRead > 0);
if (STATE_UPDATER.get(this) == State.Closed) {
if (isClosed()) {
callback.readEntriesFailed(new CursorAlreadyClosedException("Cursor was already closed"), ctx);
return;
}
Expand Down Expand Up @@ -628,6 +629,10 @@ public void asyncReadEntriesOrWait(int numberOfEntriesToRead, ReadEntriesCallbac
}
}

private boolean isClosed() {
return state == State.Closed || state == State.Closing;
}

@Override
public boolean cancelPendingReadRequest() {
if (log.isDebugEnabled()) {
Expand Down Expand Up @@ -1351,7 +1356,7 @@ public void asyncMarkDelete(final Position position, Map<String, Long> propertie
checkNotNull(position);
checkArgument(position instanceof PositionImpl);

if (STATE_UPDATER.get(this) == State.Closed) {
if (isClosed()) {
callback.markDeleteFailed(new ManagedLedgerException("Cursor was already closed"), ctx);
return;
}
Expand Down Expand Up @@ -1567,7 +1572,7 @@ public void deleteFailed(ManagedLedgerException exception, Object ctx) {

@Override
public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallback callback, Object ctx) {
if (state == State.Closed) {
if (isClosed()) {
callback.deleteFailed(new ManagedLedgerException("Cursor was already closed"), ctx);
return;
}
Expand Down Expand Up @@ -1885,6 +1890,14 @@ private boolean shouldPersistUnackRangesToLedger() {

private void persistPositionMetaStore(long cursorsLedgerId, PositionImpl position, Map<String, Long> properties,
MetaStoreCallback<Void> callback, boolean persistIndividualDeletedMessageRanges) {
if (state == State.Closed) {
ledger.getExecutor().execute(safeRun(() -> {
callback.operationFailed(new MetaStoreException(
new ManagedLedgerException.CursorAlreadyClosedException(name + " cursor already closed")));
}));
return;
}

// When closing we store the last mark-delete position in the z-node itself, so we won't need the cursor ledger,
// hence we write it as -1. The cursor ledger is deleted once the z-node write is confirmed.
ManagedCursorInfo.Builder info = ManagedCursorInfo.newBuilder() //
Expand Down Expand Up @@ -1919,13 +1932,14 @@ public void operationFailed(MetaStoreException e) {

@Override
public void asyncClose(final AsyncCallbacks.CloseCallback callback, final Object ctx) {
State oldState = STATE_UPDATER.getAndSet(this, State.Closed);
if (oldState == State.Closed) {
State oldState = STATE_UPDATER.getAndSet(this, State.Closing);
if (oldState == State.Closed || oldState == State.Closing) {
log.info("[{}] [{}] State is already closed", ledger.getName(), name);
callback.closeComplete(ctx);
return;
}
persistPosition(-1, lastMarkDeleteEntry.newPosition, lastMarkDeleteEntry.properties, callback, ctx);
STATE_UPDATER.set(this, State.Closed);
}

/**
Expand Down Expand Up @@ -2175,7 +2189,7 @@ boolean shouldCloseLedger(LedgerHandle lh) {
long now = clock.millis();
if ((lh.getLastAddConfirmed() >= config.getMetadataMaxEntriesPerLedger()
|| lastLedgerSwitchTimestamp < (now - config.getLedgerRolloverTimeout() * 1000))
&& STATE_UPDATER.get(this) != State.Closed) {
&& (STATE_UPDATER.get(this) != State.Closed && STATE_UPDATER.get(this) != State.Closing)) {
// It's safe to modify the timestamp since this method will be only called from a callback, implying that
// calls will be serialized on one single thread
lastLedgerSwitchTimestamp = now;
Expand Down