codelipenghui commented on code in PR #4901:
URL: https://github.com/apache/bookkeeper/pull/4901#discussion_r4262690779
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java:
##########
@@ -240,7 +240,57 @@ public ByteBuf getEntry(long ledgerId, long entryId)
throws IOException, BookieE
@Override
public long getLastAddConfirmed(long ledgerId) throws IOException {
- return interleavedLedgerStorage.getLastAddConfirmed(ledgerId);
+ LedgerCache ledgerCache = interleavedLedgerStorage.getLedgerCache();
+ Long cachedLac;
+ try {
+ cachedLac = ledgerCache.getLastAddConfirmed(ledgerId);
+ } catch (Bookie.NoLedgerException e) {
+ cachedLac = null;
+ }
+ if (cachedLac != null) {
+ return cachedLac;
+ }
+
+ Long memTableLac = lastAddConfirmedFromMemTable(ledgerId);
+ if (memTableLac != null) {
+ try {
+ return ledgerCache.updateLastAddConfirmed(ledgerId,
memTableLac);
+ } catch (Bookie.NoLedgerException e) {
+ return memTableLac;
+ }
+ }
+
+ try {
+ return interleavedLedgerStorage.getLastAddConfirmed(ledgerId);
+ } catch (Bookie.NoLedgerException | Bookie.NoEntryException e) {
+ // Recheck after the miss because the memtable can flush while the
index is read.
Review Comment:
Nit: with the new order, this comment no longer describes the case. The
index is only read after the memtable had nothing for this ledger, and
unflushed entries stay visible in the memtable until all of them are in the
index, so a flush can't cause this miss. The recheck now only matters if a
first entry is added after the memtable check above. Maybe: `// An entry may
have been added after the memtable check above.`
##########
bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SortedLedgerStorageTest.java:
##########
@@ -191,4 +197,139 @@ public void testGetListOfEntriesOfLedgerAfterFlush()
throws IOException {
}));
}
}
+
+ /**
+ * A newer memtable LAC must remain visible after an evicted FileInfo is
flushed.
+ */
+ @Test
+ public void testGetLastAddConfirmedAfterFileInfoEviction() throws
Exception {
+ File tmpDir = File.createTempFile("bkLac", ".dir");
+ tmpDir.delete();
+ tmpDir.mkdir();
+
BookieImpl.checkDirectoryStructure(BookieImpl.getCurrentDirectory(tmpDir));
+
+ ServerConfiguration limited =
TestBKConfiguration.newServerConfiguration();
+ limited.setEntryLogPerLedgerEnabled(conf.isEntryLogPerLedgerEnabled());
+ limited.setLedgerDirNames(new String[] { tmpDir.toString() });
+ limited.setOpenFileLimit(2);
+ limited.setNumAddWorkerThreads(1);
+ limited.setNumReadWorkerThreads(1);
+ limited.setSkipListSizeLimit(64 * 1024 * 1024);
+
+ LedgerDirsManager dirs = new LedgerDirsManager(limited,
limited.getLedgerDirs(),
+ new DiskChecker(limited.getDiskUsageThreshold(),
limited.getDiskUsageWarnThreshold()));
+ InterleavedLedgerStorage underlying = new InterleavedLedgerStorage();
+ SortedLedgerStorage storage = new SortedLedgerStorage(underlying);
+ storage.initialize(limited, null, dirs, dirs,
+ statsProvider.getStatsLogger(BOOKIE_SCOPE),
UnpooledByteBufAllocator.DEFAULT);
+ storage.setCheckpointSource(CheckpointSource.DEFAULT);
+ storage.setCheckpointer(checkpointer);
+ try {
+ long ledgerId = 0L;
+ storage.setMasterKey(ledgerId, ("ledger-" + ledgerId).getBytes());
+ for (long entryId = 0; entryId < 5; entryId++) {
+ ByteBuf entry = Unpooled.buffer(128);
+ entry.writeLong(ledgerId);
+ entry.writeLong(entryId);
+ entry.writeLong(entryId - 1);
+ entry.writeByte(1);
+ storage.addEntry(entry);
+ }
+ storage.flush();
+
+ for (long entryId = 5; entryId < 10; entryId++) {
+ ByteBuf entry = Unpooled.buffer(128);
+ entry.writeLong(ledgerId);
+ entry.writeLong(entryId);
+ entry.writeLong(entryId - 1);
+ entry.writeByte(1);
+ storage.addEntry(entry);
+ }
+ LedgerCacheImpl ledgerCache = (LedgerCacheImpl)
underlying.ledgerCache;
+ IndexPersistenceMgr indexPersistenceManager =
ledgerCache.getIndexPersistenceManager();
+ indexPersistenceManager.readFileInfoCache.invalidate(ledgerId);
+ indexPersistenceManager.writeFileInfoCache.invalidate(ledgerId);
+ indexPersistenceManager.readFileInfoCache.cleanUp();
+ indexPersistenceManager.writeFileInfoCache.cleanUp();
+
+ assertEquals("LAC of an evicted ledger must include newer memtable
entries",
+ 8L, storage.getLastAddConfirmed(ledgerId));
+ storage.flush();
+ assertTrue("memtable must be empty after a real flush",
storage.memTable.isEmpty());
+ assertEquals("LAC must not move backwards after the memtable is
flushed",
+ 8L, storage.getLastAddConfirmed(ledgerId));
+ assertThrows(Bookie.NoLedgerException.class, () ->
storage.getLastAddConfirmed(9_999L));
+ } finally {
+ storage.shutdown();
+ }
+ }
+
+ @Test
+ public void testGetLastAddConfirmedRetriesAfterIndexMiss() throws
Exception {
+ InterleavedLedgerStorage underlying =
mock(InterleavedLedgerStorage.class);
+ LedgerCache ledgerCache = mock(LedgerCache.class);
+ when(underlying.getLedgerCache()).thenReturn(ledgerCache);
+ when(ledgerCache.getLastAddConfirmed(0L)).thenReturn(null);
+ when(underlying.getLastAddConfirmed(0L))
+ .thenThrow(new Bookie.NoEntryException(0L, -1L))
+ .thenReturn(1L);
+ SortedLedgerStorage storage = new SortedLedgerStorage(underlying);
+ storage.initialize(conf, null, ledgerDirsManager, ledgerDirsManager,
+ statsProvider.getStatsLogger(BOOKIE_SCOPE),
UnpooledByteBufAllocator.DEFAULT);
+ storage.setCheckpointSource(checkpointSource);
+ storage.setCheckpointer(checkpointer);
+ try {
+ assertEquals(1L, storage.getLastAddConfirmed(0L));
+ verify(underlying, times(2)).getLastAddConfirmed(0L);
+ } finally {
+ storage.shutdown();
+ }
+ }
+
+ @Test
+ public void testGetLastAddConfirmedUsesCachedValueBeforeMemTable() throws
Exception {
+ InterleavedLedgerStorage underlying =
mock(InterleavedLedgerStorage.class);
+ LedgerCache ledgerCache = mock(LedgerCache.class);
+ when(underlying.getLedgerCache()).thenReturn(ledgerCache);
+ when(ledgerCache.getLastAddConfirmed(0L)).thenReturn(8L);
+ SortedLedgerStorage storage = new SortedLedgerStorage(underlying);
+ storage.initialize(conf, null, ledgerDirsManager, ledgerDirsManager,
+ statsProvider.getStatsLogger(BOOKIE_SCOPE),
UnpooledByteBufAllocator.DEFAULT);
+ storage.setCheckpointSource(checkpointSource);
+ storage.setCheckpointer(checkpointer);
+ EntryMemTable memTable = mock(EntryMemTable.class);
+ storage.memTable = memTable;
Review Comment:
Nit: `shutdown()` now closes the mock, so the real memtable created by
`setCheckpointSource` is never closed. In the `entryLogPerLedger=true` run
that's an `EntryMemTableWithParallelFlusher`, and its 8 flush threads keep
running after the test. I counted 8 alive after `shutdown()`, and 0 without the
swap.
The class already doesn't shut down its `setUp` storage, so this is
optional. Either of these would avoid it:
- keep a reference to the original memtable and close it in `finally`;
- use `spy(storage.memTable)` with `verify(..., never()).getLastEntry(0L)`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
