Skip to content

Issue 4900: Read an unflushed ledger LAC from the memtable #4901

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

Open
kalayciburak wants to merge 8 commits into
base: master
Choose a base branch
from
+279 −1
Open

Issue 4900: Read an unflushed ledger LAC from the memtable #4901

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
0af2c62
Issue 4900: Read an unflushed ledger LAC from the memtable
kalayciburak Oct 4, 2026
11eba81
fix: prefer unflushed LAC over stale index value
kalayciburak Oct 13, 2026
4eba457
fix: preserve newer LAC after FileInfo eviction
kalayciburak Oct 13, 2026
1d1543b
fix: read cached LAC before memtable
kalayciburak Oct 13, 2026
6768dd3
test: address LAC review nits
kalayciburak Oct 13, 2026
b0a6cad
fix: reject stale LAC reads for deleted ledgers
kalayciburak Oct 15, 2026
da60ca0
fix: avoid ledger existence checks on cached LAC reads
kalayciburak Oct 21, 2026
3c64475
test: cover unflushed LAC after FileInfo eviction
kalayciburak Oct 27, 2026
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
41 changes: 40 additions & 1 deletion bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -240,7 +240,46 @@ 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();
// Throws NoLedgerException for unknown or deleted ledgers. An evicted FileInfo
// reloads from its index file without a LAC, so this returns null instead.
Long cachedLac = ledgerCache.getLastAddConfirmed(ledgerId);
if (cachedLac != null) {
return cachedLac;
}

Long memTableLac = lastAddConfirmedFromMemTable(ledgerId);
if (memTableLac != null) {
return ledgerCache.updateLastAddConfirmed(ledgerId, memTableLac);
}

try {
return interleavedLedgerStorage.getLastAddConfirmed(ledgerId);
} catch (Bookie.NoLedgerException | Bookie.NoEntryException e) {
// An entry may have been added after the memtable check above.
Long latestMemTableLac = lastAddConfirmedFromMemTable(ledgerId);
if (latestMemTableLac == null) {
return interleavedLedgerStorage.getLastAddConfirmed(ledgerId);
}
return ledgerCache.updateLastAddConfirmed(ledgerId, latestMemTableLac);
}
}

private Long lastAddConfirmedFromMemTable(long ledgerId) throws IOException {
EntryKeyValue kv = memTable.getLastEntry(ledgerId);
if (kv == null) {
return null;
}
ByteBuf entry = kv.getValueAsByteBuffer();
try {
if (entry.readableBytes() < 3 * Long.BYTES) {
return null;
}
entry.skipBytes(2 * Long.BYTES);
return entry.readLong();
} finally {
entry.release();
}
}

@Override
Expand Down
239 changes: 239 additions & 0 deletions bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SortedLedgerStorageTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,13 +23,20 @@
import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIE_SCOPE;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.buffer.UnpooledByteBufAllocator;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.PrimitiveIterator.OfLong;
Expand All @@ -41,7 +48,9 @@
import org.apache.bookkeeper.test.TestStatsProvider;
import org.apache.bookkeeper.util.DiskChecker;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

Expand All @@ -56,6 +65,9 @@ public class SortedLedgerStorageTest {
LedgerDirsManager ledgerDirsManager;
SortedLedgerStorage sortedLedgerStorage = new SortedLedgerStorage();

@Rule
public TemporaryFolder temporaryFolder = new TemporaryFolder();

final long numWrites = 2000;
final long moreNumOfWrites = 3000;
final long entriesPerWrite = 2;
Expand Down Expand Up @@ -191,4 +203,231 @@ 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);
try {
entry.writeLong(ledgerId);
entry.writeLong(entryId);
entry.writeLong(entryId - 1);
entry.writeByte(1);
storage.addEntry(entry);
} finally {
entry.release();
}
}
storage.flush();

@codelipenghui codelipenghui Oct 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This flush() doesn't flush the memtable. In this class, checkpointSource.newCheckpoint() returns Checkpoint.MAX. EntryMemTable#snapshot(Checkpoint.MAX) only takes a snapshot when kvmap.cp.compareTo(MAX) < 0, which is never true for MAX.

I checked that entry 0 is still in storage.memTable afterwards. So the next assertion runs the memtable path again, not the post-flush index path.

Using CheckpointSource.DEFAULT (which returns Checkpoint.MIN) for this storage, and asserting storage.memTable.isEmpty() after the flush, would make it test what the message says. The current code does return the right LAC from the index after a real flush.

@codelipenghui codelipenghui Oct 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since entries 0–4 are flushed here, the index lookup in getLastAddConfirmed succeeds (it returns 3), so this test only runs the try branch. The catch branch is the fix for #4900 itself: the index throws for an evicted ledger with nothing flushed. That branch isn't covered any more. The earlier version of this test was replaced, and testGetLastAddConfirmedRetriesAfterIndexMiss only reaches that branch with an empty memtable.

I checked locally with two changes to SortedLedgerStorage#getLastAddConfirmed. All 8 tests in this class still pass with either one:

Removing the write-back in the try branch does fail this test, so it's a good guard for the partially flushed case.

Could you keep this test and add a fully unflushed case next to it? With the same setup:

  1. Add entries 0–2 without flushing.
  2. Evict with the same invalidate calls.
  3. Assert LAC 1.
  4. Assert waitForLastAddConfirmedUpdate(ledgerId, 0, watcher) returns false (the write-back reached the cache).
  5. Call flush(), assert the memtable is empty, and assert LAC 1 again.

In my local checks, assertions like these fail under both changes above.


for (long entryId = 5; entryId < 10; entryId++) {
ByteBuf entry = Unpooled.buffer(128);
try {
entry.writeLong(ledgerId);
entry.writeLong(entryId);
entry.writeLong(entryId - 1);
entry.writeByte(1);
storage.addEntry(entry);
} finally {
entry.release();
}
}
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();
}
}

/**
* An unflushed LAC must be written back to the ledger cache after FileInfo eviction.
*/
@Test
public void testGetLastAddConfirmedAfterUnflushedFileInfoEviction() throws Exception {
File tmpDir = temporaryFolder.newFolder("bkLacUnflushed");
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);
boolean initialized = false;
try {
storage.initialize(limited, null, dirs, dirs,
statsProvider.getStatsLogger(BOOKIE_SCOPE), UnpooledByteBufAllocator.DEFAULT);
initialized = true;
storage.setCheckpointSource(CheckpointSource.DEFAULT);
storage.setCheckpointer(checkpointer);
long ledgerId = 0L;
storage.setMasterKey(ledgerId, ("ledger-" + ledgerId).getBytes());
for (long entryId = 0; entryId < 3; entryId++) {
ByteBuf entry = Unpooled.buffer(128);
try {
entry.writeLong(ledgerId);
entry.writeLong(entryId);
entry.writeLong(entryId - 1);
entry.writeByte(1);
storage.addEntry(entry);
} finally {
entry.release();
}
}
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 must include unflushed memtable entries", 1L, storage.getLastAddConfirmed(ledgerId));
assertFalse("LAC write-back must make the update immediately visible to long polls",
storage.waitForLastAddConfirmedUpdate(ledgerId, 0L, notification -> { }));
storage.flush();
assertTrue("memtable must be empty after a real flush", storage.memTable.isEmpty());
assertEquals("LAC must remain visible after the memtable is flushed",
1L, storage.getLastAddConfirmed(ledgerId));
} finally {
if (initialized) {
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);
verify(underlying, times(0)).ledgerExists(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 = spy(storage.memTable);
storage.memTable = memTable;
try {
assertEquals(8L, storage.getLastAddConfirmed(0L));
verify(memTable, times(0)).getLastEntry(0L);
verify(underlying, times(0)).getLastAddConfirmed(0L);
verify(underlying, times(0)).ledgerExists(0L);
} finally {
storage.shutdown();
}
}

@Test
public void testGetLastAddConfirmedFallsBackToMemTableWhenCacheMisses() throws Exception {
InterleavedLedgerStorage underlying = mock(InterleavedLedgerStorage.class);
LedgerCache ledgerCache = mock(LedgerCache.class);
when(underlying.getLedgerCache()).thenReturn(ledgerCache);
when(ledgerCache.getLastAddConfirmed(0L)).thenReturn(null);
when(ledgerCache.updateLastAddConfirmed(0L, 8L)).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);
try {
ByteBuffer entry = ByteBuffer.allocate(3 * Long.BYTES);
entry.putLong(0L).putLong(9L).putLong(8L).flip();
storage.memTable.addEntry(0L, 9L, entry, storage);

assertEquals(8L, storage.getLastAddConfirmed(0L));
verify(ledgerCache).updateLastAddConfirmed(0L, 8L);
verify(underlying, times(0)).getLastAddConfirmed(0L);
verify(underlying, times(0)).ledgerExists(0L);
} finally {
storage.shutdown();
}
}

@Test
public void testGetLastAddConfirmedDoesNotReadDeletedLedgerFromMemTable() throws Exception {
long ledgerId = 0L;
sortedLedgerStorage.setMasterKey(ledgerId, ("ledger-" + ledgerId).getBytes());
ByteBuf entry = Unpooled.buffer(128);
entry.writeLong(ledgerId);
entry.writeLong(0L);
entry.writeLong(-1L);
entry.writeByte(1);
try {
sortedLedgerStorage.addEntry(entry);
sortedLedgerStorage.deleteLedger(ledgerId);
assertThrows(Bookie.NoLedgerException.class,
() -> sortedLedgerStorage.getLastAddConfirmed(ledgerId));
} finally {
entry.release();
}
}
}