diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java index 4981910ce9b84..e611d634f9fb4 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java @@ -151,7 +151,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback { protected Map propertiesMap; protected final MetaStore store; - private final ConcurrentLongHashMap> ledgerCache = new ConcurrentLongHashMap<>( + final ConcurrentLongHashMap> ledgerCache = new ConcurrentLongHashMap<>( 16 /* initial capacity */, 1 /* number of sections */); protected final NavigableMap ledgers = new ConcurrentSkipListMap<>(); private volatile Stat ledgersStat; @@ -2410,7 +2410,7 @@ void internalTrimLedgers(boolean isTruncate, CompletableFuture promise) { if (log.isDebugEnabled()) { log.debug("[{}] Ledger {} not deleted. Neither expired nor over-quota", name, ls.getLedgerId()); } - break; + invalidateReadHandle(ls.getLedgerId()); } } diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java index 25c1725bdb4cc..cd805379ac059 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java @@ -3097,4 +3097,80 @@ public void testOpEntryAdd_toString_doesNotThrowNPE(){ ", dataLength=" + dataLength + '}'; } + + @Test + public void testInvalidateReadHandleWhenDeleteLedger() throws Exception { + ManagedLedgerConfig config = new ManagedLedgerConfig(); + config.setMaxEntriesPerLedger(1); + + // Verify the read handle should be invalidated after ledger been removed. + ManagedLedgerImpl ledger = (ManagedLedgerImpl)factory.open("testInvalidateReadHandleWhenDeleteLedger", config); + ManagedCursor cursor = ledger.openCursor("test-cursor"); + ManagedCursor cursor2 = ledger.openCursor("test-cursor2"); + final int entries = 3; + for (int i = 0; i < entries; i++) { + ledger.addEntry(String.valueOf(i).getBytes(Encoding)); + } + List entryList = cursor.readEntries(3); + assertEquals(entryList.size(), 3); + assertEquals(ledger.ledgers.size(), 3); + assertEquals(ledger.ledgerCache.size(), 2); + cursor.clearBacklog(); + cursor2.clearBacklog(); + ledger.trimConsumedLedgersInBackground(Futures.NULL_PROMISE); + Awaitility.await().untilAsserted(() -> { + assertEquals(ledger.ledgers.size(), 1); + assertEquals(ledger.ledgerCache.size(), 0); + }); + + cursor.close(); + cursor2.close(); + ledger.close(); + } + + @Test + public void testInvalidateReadHandleWhenConsumed() throws Exception { + ManagedLedgerConfig config = new ManagedLedgerConfig(); + config.setMaxEntriesPerLedger(1); + // Verify the read handle should be invalidated when all cursors consumed + // even if the ledger can not been removed due to the data retention + config.setRetentionSizeInMB(50); + config.setRetentionTime(1, TimeUnit.DAYS); + ManagedLedgerImpl ledger = (ManagedLedgerImpl)factory.open("testInvalidateReadHandleWhenConsumed", config); + ManagedCursor cursor = ledger.openCursor("test-cursor"); + ManagedCursor cursor2 = ledger.openCursor("test-cursor2"); + final int entries = 3; + for (int i = 0; i < entries; i++) { + ledger.addEntry(String.valueOf(i).getBytes(Encoding)); + } + List entryList = cursor.readEntries(3); + assertEquals(entryList.size(), 3); + assertEquals(ledger.ledgers.size(), 3); + assertEquals(ledger.ledgerCache.size(), 2); + cursor.clearBacklog(); + cursor2.clearBacklog(); + ledger.trimConsumedLedgersInBackground(Futures.NULL_PROMISE); + Awaitility.await().untilAsserted(() -> { + assertEquals(ledger.ledgers.size(), 3); + assertEquals(ledger.ledgerCache.size(), 0); + }); + + // Verify the ReadHandle can be reopened. + ManagedCursor cursor3 = ledger.openCursor("test-cursor3", InitialPosition.Earliest); + entryList = cursor3.readEntries(3); + assertEquals(entryList.size(), 3); + assertEquals(ledger.ledgerCache.size(), 2); + cursor3.clearBacklog(); + ledger.trimConsumedLedgersInBackground(Futures.NULL_PROMISE); + Awaitility.await().untilAsserted(() -> { + assertEquals(ledger.ledgers.size(), 3); + assertEquals(ledger.ledgerCache.size(), 0); + }); + + + cursor.close(); + cursor2.close(); + cursor3.close(); + ledger.close(); + } }