codelipenghui commented on code in PR #4901:
URL: https://github.com/apache/bookkeeper/pull/4901#discussion_r4260813570
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java:
##########
@@ -240,7 +240,39 @@ public ByteBuf getEntry(long ledgerId, long entryId)
throws IOException, BookieE
@Override
public long getLastAddConfirmed(long ledgerId) throws IOException {
- return interleavedLedgerStorage.getLastAddConfirmed(ledgerId);
+ // The index can still expose the committed LAC while a newer entry is
in the memtable.
+ Long memTableLac = lastAddConfirmedFromMemTable(ledgerId);
+ try {
+ long persistedLac =
interleavedLedgerStorage.getLastAddConfirmed(ledgerId);
+ if (memTableLac == null) {
+ return persistedLac;
+ }
+ return Math.max(persistedLac, memTableLac);
Review Comment:
The returned value is now correct while the newer entries are still in the
memtable. But `interleavedLedgerStorage.getLastAddConfirmed()` still caches the
index-derived LAC in the reloaded `FileInfo`
(`InterleavedLedgerStorage#getLastAddConfirmed` calls
`ledgerCache.updateLastAddConfirmed`). Nothing writes the memtable value back.
Repro, using a real `SortedLedgerStorage` with `CheckpointSource.DEFAULT` so
`flush()` really flushes:
1. Add entries 0–4 (LAC = entryId − 1) and flush. Add entries 5–9. Evict
ledger 0's FileInfo.
2. `getLastAddConfirmed(0)` returns 8, which is correct. But the ledger
cache now holds 3.
3. Call `flush()` while the FileInfo is still cached.
`getLastAddConfirmed(0)` now returns **3**.
So once the rest of the ledger is flushed, the LAC still goes backwards (8 →
3), which is the case @djsweet described. It stays at 3 until a newer LAC
arrives or the FileInfo is evicted again.
Writing the winning value back fixes it. `FileInfo#setLastAddConfirmed` only
moves forward and also notifies long-poll watchers:
```java
if (memTableLac == null || memTableLac <= persistedLac) {
return persistedLac;
}
return interleavedLedgerStorage.ledgerCache.updateLastAddConfirmed(ledgerId,
memTableLac);
```
The catch branch needs the same write-back for `latestMemTableLac`, ignoring
`NoLedgerException`. With that, the repro returns 8 both before and after the
flush. Alternatively, check the cached LAC first (see the next comment).
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java:
##########
@@ -240,7 +240,39 @@ public ByteBuf getEntry(long ledgerId, long entryId)
throws IOException, BookieE
@Override
public long getLastAddConfirmed(long ledgerId) throws IOException {
- return interleavedLedgerStorage.getLastAddConfirmed(ledgerId);
+ // The index can still expose the committed LAC while a newer entry is
in the memtable.
+ Long memTableLac = lastAddConfirmedFromMemTable(ledgerId);
Review Comment:
This puts a memtable lookup in front of every call: a read lock, up to two
skip-list `floorKey` calls, and a get-entry stats event in
`EntryMemTable#getLastEntry`. That includes the common case where the FileInfo
already has the LAC cached, and `ReadEntryProcessorV3` calls this on every V3
entry read.
I ran a quick local loop: 1000 ledgers × 200 unflushed entries, LAC cached.
- Single thread: about 38 ns per call on master, about 160 ns with this
change.
- 8 threads: about 8.5 M calls/s on master, about 3.9 M with this change.
Every LAC read also counts as a memtable get in the skip-list stats.
Checking the cache first keeps the common path the same as master and only
touches the memtable after an eviction. This is essentially the order #4902
uses:
```java
LedgerCache ledgerCache = interleavedLedgerStorage.ledgerCache;
Long lac = null;
try {
lac = ledgerCache.getLastAddConfirmed(ledgerId);
} catch (Bookie.NoLedgerException nle) {
// no FileInfo/index; the memtable may still have the ledger
}
if (lac != null) {
return lac;
}
Long memLac = lastAddConfirmedFromMemTable(ledgerId);
if (memLac != null) {
try {
return ledgerCache.updateLastAddConfirmed(ledgerId, memLac);
} catch (Bookie.NoLedgerException nle) {
return memLac;
}
}
// nothing unflushed (or it was just flushed): the index is complete
return interleavedLedgerStorage.getLastAddConfirmed(ledgerId);
```
This is safe against a concurrent flush. `flushSnapshot` writes every
snapshot entry to the index before `clearSnapshot`, and `getLastEntry` still
sees the snapshot until then. So if the memtable has nothing, the index already
has everything, and the retry is no longer needed.
Locally this passes the eviction test, the partially flushed repro above and
an end-to-end V3 read, at the same per-call cost as master.
##########
bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SortedLedgerStorageTest.java:
##########
@@ -191,4 +196,101 @@ public void testGetListOfEntriesOfLedgerAfterFlush()
throws IOException {
}));
}
}
+
+ /**
+ * An unflushed ledger's LAC lives in a FileInfo cache entry. Once that
entry is
+ * evicted, the index file still does not exist, so the LAC has to come
from the memtable.
+ */
+ @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()));
+ SortedLedgerStorage storage = new SortedLedgerStorage();
+ storage.initialize(limited, null, dirs, dirs,
+ statsProvider.getStatsLogger(BOOKIE_SCOPE),
UnpooledByteBufAllocator.DEFAULT);
+ storage.setCheckpointSource(checkpointSource);
+ storage.setCheckpointer(checkpointer);
+ try {
+ long lac = 3L;
+ int ledgers = 12;
+ for (long ledgerId = 0; ledgerId < ledgers; ledgerId++) {
+ storage.setMasterKey(ledgerId, ("ledger-" +
ledgerId).getBytes());
+ ByteBuf entry = Unpooled.buffer(128);
+ entry.writeLong(ledgerId);
+ entry.writeLong(0L);
+ entry.writeLong(lac);
+ entry.writeByte(1);
+ storage.addEntry(entry);
+ }
+ assertEquals("LAC of an unflushed ledger after file-info eviction",
+ lac, storage.getLastAddConfirmed(0L));
+ storage.flush();
Review Comment:
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.
##########
bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SortedLedgerStorageTest.java:
##########
@@ -191,4 +196,101 @@ public void testGetListOfEntriesOfLedgerAfterFlush()
throws IOException {
}));
}
}
+
+ /**
+ * An unflushed ledger's LAC lives in a FileInfo cache entry. Once that
entry is
+ * evicted, the index file still does not exist, so the LAC has to come
from the memtable.
+ */
+ @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()));
+ SortedLedgerStorage storage = new SortedLedgerStorage();
+ storage.initialize(limited, null, dirs, dirs,
+ statsProvider.getStatsLogger(BOOKIE_SCOPE),
UnpooledByteBufAllocator.DEFAULT);
+ storage.setCheckpointSource(checkpointSource);
+ storage.setCheckpointer(checkpointer);
+ try {
+ long lac = 3L;
+ int ledgers = 12;
+ for (long ledgerId = 0; ledgerId < ledgers; ledgerId++) {
+ storage.setMasterKey(ledgerId, ("ledger-" +
ledgerId).getBytes());
+ ByteBuf entry = Unpooled.buffer(128);
+ entry.writeLong(ledgerId);
+ entry.writeLong(0L);
+ entry.writeLong(lac);
+ entry.writeByte(1);
+ storage.addEntry(entry);
+ }
+ assertEquals("LAC of an unflushed ledger after file-info eviction",
+ lac, storage.getLastAddConfirmed(0L));
+ storage.flush();
+ assertEquals("LAC still readable after the memtable is flushed",
+ lac, storage.getLastAddConfirmed(0L));
+ try {
+ storage.getLastAddConfirmed(9_999L);
+ assertTrue("missing ledger should not report a LAC", false);
+ } catch (Bookie.NoLedgerException | Bookie.NoEntryException
expected) {
+ // index has no file and the memtable has no entry
+ }
Review Comment:
Nit: `assertThrows(IOException.class, () ->
storage.getLastAddConfirmed(9_999L))` reads better than `assertTrue(msg,
false)`.
Also, entry 0 can't carry LAC 3 (`lac = 3L` above), because a piggybacked
LAC is always lower than the entry id. Writing a few entries with `LAC =
entryId - 1` would keep the test realistic.
##########
bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SortedLedgerStorageTest.java:
##########
@@ -191,4 +196,101 @@ public void testGetListOfEntriesOfLedgerAfterFlush()
throws IOException {
}));
}
}
+
+ /**
+ * An unflushed ledger's LAC lives in a FileInfo cache entry. Once that
entry is
+ * evicted, the index file still does not exist, so the LAC has to come
from the memtable.
+ */
+ @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()));
+ SortedLedgerStorage storage = new SortedLedgerStorage();
+ storage.initialize(limited, null, dirs, dirs,
+ statsProvider.getStatsLogger(BOOKIE_SCOPE),
UnpooledByteBufAllocator.DEFAULT);
+ storage.setCheckpointSource(checkpointSource);
+ storage.setCheckpointer(checkpointer);
+ try {
+ long lac = 3L;
+ int ledgers = 12;
+ for (long ledgerId = 0; ledgerId < ledgers; ledgerId++) {
+ storage.setMasterKey(ledgerId, ("ledger-" +
ledgerId).getBytes());
+ ByteBuf entry = Unpooled.buffer(128);
+ entry.writeLong(ledgerId);
+ entry.writeLong(0L);
+ entry.writeLong(lac);
+ entry.writeByte(1);
+ storage.addEntry(entry);
+ }
+ assertEquals("LAC of an unflushed ledger after file-info eviction",
+ lac, storage.getLastAddConfirmed(0L));
+ storage.flush();
+ assertEquals("LAC still readable after the memtable is flushed",
+ lac, storage.getLastAddConfirmed(0L));
+ try {
+ storage.getLastAddConfirmed(9_999L);
+ assertTrue("missing ledger should not report a LAC", false);
+ } catch (Bookie.NoLedgerException | Bookie.NoEntryException
expected) {
+ // index has no file and the memtable has no entry
+ }
+ } finally {
+ storage.shutdown();
+ }
+ }
+
+ /**
+ * A committed LAC can be stale while a newer entry is still in the
memtable.
+ */
+ @Test
+ public void testGetLastAddConfirmedPrefersMemTableForExistingLedger()
throws Exception {
+ InterleavedLedgerStorage underlying =
mock(InterleavedLedgerStorage.class);
+ when(underlying.getLastAddConfirmed(0L)).thenReturn(0L);
Review Comment:
Mocking `InterleavedLedgerStorage` hides the side effect behind the
remaining bug. The real `getLastAddConfirmed()` also stores the index-derived
LAC in the FileInfo cache, and that is what later comes back as the stale
value. So this test passes while a real storage still goes 8 → 3 (see the
comment on `SortedLedgerStorage`). Both mock tests would also need rework with
either fix, because a mock's `ledgerCache` field is `null`.
Could this be a real-storage test instead? Roughly:
1. Use `CheckpointSource.DEFAULT` so `flush()` really flushes.
2. Add entries 0–4 and flush, then add 5–9.
3. Evict ledger 0's FileInfo, either with `openFileLimit` pressure as in the
test above or by invalidating both `IndexPersistenceMgr` caches from the test
package.
4. Assert 8, call `flush()`, and assert 8 again.
The last assertion fails on the current head.
##########
bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SortedLedgerStorageTest.java:
##########
@@ -191,4 +196,101 @@ public void testGetListOfEntriesOfLedgerAfterFlush()
throws IOException {
}));
}
}
+
+ /**
+ * An unflushed ledger's LAC lives in a FileInfo cache entry. Once that
entry is
+ * evicted, the index file still does not exist, so the LAC has to come
from the memtable.
Review Comment:
Nit: the index file does exist after eviction.
`FileInfoBackingCache#releaseFileInfo` calls `FileInfo.close(true)`, which
creates a header-only `.idx` (1024 bytes). What's missing are the index
entries. That's why the lookup throws `NoEntryException`, and why master
returns `ENOENTRY` here rather than `ENOLEDGER`.
--
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]
