@@ -166,6 +166,7 @@ enum State {
166166 NoLedger , // There is no metadata ledger open for writing
167167 Open , // Metadata ledger is ready
168168 SwitchingLedger , // The metadata ledger is being switched
169+ Closing , // The managed cursor is closing
169170 Closed // The managed cursor has been closed
170171 }
171172
@@ -282,6 +283,15 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac
282283
283284 // Read the last entry in the ledger
284285 long lastEntryInLedger = lh .getLastAddConfirmed ();
286+
287+ if (lastEntryInLedger < 0 ) {
288+ log .warn ("[{}] Error reading from metadata ledger {} for consumer {}: No entries in ledger" ,
289+ ledger .getName (), ledgerId , name );
290+ // Rewind to last cursor snapshot available
291+ initialize (getRollbackPosition (info ), callback );
292+ return ;
293+ }
294+
285295 lh .asyncReadEntries (lastEntryInLedger , lastEntryInLedger , (rc1 , lh1 , seq , ctx1 ) -> {
286296 if (log .isDebugEnabled ()) {
287297 log .debug ("[{}} readComplete rc={} entryId={}" , ledger .getName (), rc1 , lh1 .getLastAddConfirmed ());
@@ -430,7 +440,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
430440 public void asyncReadEntries (final int numberOfEntriesToRead , final ReadEntriesCallback callback ,
431441 final Object ctx ) {
432442 checkArgument (numberOfEntriesToRead > 0 );
433- if (STATE_UPDATER . get ( this ) == State . Closed ) {
443+ if (isClosed () ) {
434444 callback .readEntriesFailed (new ManagedLedgerException ("Cursor was already closed" ), ctx );
435445 return ;
436446 }
@@ -480,7 +490,7 @@ public void readEntryComplete(Entry entry, Object ctx) {
480490 public void asyncGetNthEntry (int n , IndividualDeletedEntries deletedEntries , ReadEntryCallback callback ,
481491 Object ctx ) {
482492 checkArgument (n > 0 );
483- if (STATE_UPDATER . get ( this ) == State . Closed ) {
493+ if (isClosed () ) {
484494 callback .readEntryFailed (new ManagedLedgerException ("Cursor was already closed" ), ctx );
485495 return ;
486496 }
@@ -545,7 +555,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
545555 @ Override
546556 public void asyncReadEntriesOrWait (int numberOfEntriesToRead , ReadEntriesCallback callback , Object ctx ) {
547557 checkArgument (numberOfEntriesToRead > 0 );
548- if (STATE_UPDATER . get ( this ) == State . Closed ) {
558+ if (isClosed () ) {
549559 callback .readEntriesFailed (new CursorAlreadyClosedException ("Cursor was already closed" ), ctx );
550560 return ;
551561 }
@@ -619,6 +629,10 @@ public void asyncReadEntriesOrWait(int numberOfEntriesToRead, ReadEntriesCallbac
619629 }
620630 }
621631
632+ private boolean isClosed () {
633+ return state == State .Closed || state == State .Closing ;
634+ }
635+
622636 @ Override
623637 public boolean cancelPendingReadRequest () {
624638 if (log .isDebugEnabled ()) {
@@ -1342,7 +1356,7 @@ public void asyncMarkDelete(final Position position, Map<String, Long> propertie
13421356 checkNotNull (position );
13431357 checkArgument (position instanceof PositionImpl );
13441358
1345- if (STATE_UPDATER . get ( this ) == State . Closed ) {
1359+ if (isClosed () ) {
13461360 callback .markDeleteFailed (new ManagedLedgerException ("Cursor was already closed" ), ctx );
13471361 return ;
13481362 }
@@ -1558,7 +1572,7 @@ public void deleteFailed(ManagedLedgerException exception, Object ctx) {
15581572
15591573 @ Override
15601574 public void asyncDelete (Iterable <Position > positions , AsyncCallbacks .DeleteCallback callback , Object ctx ) {
1561- if (state == State . Closed ) {
1575+ if (isClosed () ) {
15621576 callback .deleteFailed (new ManagedLedgerException ("Cursor was already closed" ), ctx );
15631577 return ;
15641578 }
@@ -1876,6 +1890,14 @@ private boolean shouldPersistUnackRangesToLedger() {
18761890
18771891 private void persistPositionMetaStore (long cursorsLedgerId , PositionImpl position , Map <String , Long > properties ,
18781892 MetaStoreCallback <Void > callback , boolean persistIndividualDeletedMessageRanges ) {
1893+ if (state == State .Closed ) {
1894+ ledger .getExecutor ().execute (safeRun (() -> {
1895+ callback .operationFailed (new MetaStoreException (
1896+ new ManagedLedgerException .CursorAlreadyClosedException (name + " cursor already closed" )));
1897+ }));
1898+ return ;
1899+ }
1900+
18791901 // When closing we store the last mark-delete position in the z-node itself, so we won't need the cursor ledger,
18801902 // hence we write it as -1. The cursor ledger is deleted once the z-node write is confirmed.
18811903 ManagedCursorInfo .Builder info = ManagedCursorInfo .newBuilder () //
@@ -1910,13 +1932,14 @@ public void operationFailed(MetaStoreException e) {
19101932
19111933 @ Override
19121934 public void asyncClose (final AsyncCallbacks .CloseCallback callback , final Object ctx ) {
1913- State oldState = STATE_UPDATER .getAndSet (this , State .Closed );
1914- if (oldState == State .Closed ) {
1935+ State oldState = STATE_UPDATER .getAndSet (this , State .Closing );
1936+ if (oldState == State .Closed || oldState == State . Closing ) {
19151937 log .info ("[{}] [{}] State is already closed" , ledger .getName (), name );
19161938 callback .closeComplete (ctx );
19171939 return ;
19181940 }
19191941 persistPosition (-1 , lastMarkDeleteEntry .newPosition , lastMarkDeleteEntry .properties , callback , ctx );
1942+ STATE_UPDATER .set (this , State .Closed );
19201943 }
19211944
19221945 /**
@@ -2059,7 +2082,7 @@ public void deleteComplete(int rc, Object ctx) {
20592082 });
20602083 }));
20612084 }, Collections .emptyMap ());
2062-
2085+
20632086 }
20642087
20652088 private List <LongProperty > buildPropertiesMap (Map <String , Long > properties ) {
@@ -2166,7 +2189,7 @@ boolean shouldCloseLedger(LedgerHandle lh) {
21662189 long now = clock .millis ();
21672190 if ((lh .getLastAddConfirmed () >= config .getMetadataMaxEntriesPerLedger ()
21682191 || lastLedgerSwitchTimestamp < (now - config .getLedgerRolloverTimeout () * 1000 ))
2169- && STATE_UPDATER .get (this ) != State .Closed ) {
2192+ && ( STATE_UPDATER .get (this ) != State .Closed && STATE_UPDATER . get ( this ) != State . Closing ) ) {
21702193 // It's safe to modify the timestamp since this method will be only called from a callback, implying that
21712194 // calls will be serialized on one single thread
21722195 lastLedgerSwitchTimestamp = now ;
0 commit comments