Copilot commented on code in PR #4740:
URL: https://github.com/apache/bookkeeper/pull/4740#discussion_r3048747327
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java:
##########
@@ -776,6 +788,15 @@ public void checkpointComplete(Checkpoint checkpoint,
boolean compact) throws IO
LogMarkCheckpoint lmcheckpoint = (LogMarkCheckpoint) checkpoint;
LastLogMark mark = lmcheckpoint.mark;
+ // Monotonic check: only advance lastMark forward, never backwards.
+ synchronized (checkpointLock) {
+ if (mark.getCurMark().compare(lastPersistedMark) <= 0) {
+ return;
+ }
+ lastPersistedMark.setLogMark(
+ mark.getCurMark().getLogFileId(),
mark.getCurMark().getLogFileOffset());
+ }
+
mark.rollLog(mark);
Review Comment:
lastPersistedMark is advanced before rollLog() and before any I/O/GC occurs.
If rollLog() throws (e.g., NoWritableLedgerDirException) or fails to persist
the file, subsequent checkpointComplete calls with the same/newer mark can be
skipped, leaving lastMark on disk stale. Update lastPersistedMark only after
successfully persisting lastMark (or keep it consistent by performing the
monotonic decision + persistence atomically under the lock).
```suggestion
// Keep the monotonic decision and persistence consistent by only
// updating lastPersistedMark after rollLog succeeds.
synchronized (checkpointLock) {
if (mark.getCurMark().compare(lastPersistedMark) <= 0) {
return;
}
mark.rollLog(mark);
lastPersistedMark.setLogMark(
mark.getCurMark().getLogFileId(),
mark.getCurMark().getLogFileOffset());
}
```
##########
bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java:
##########
@@ -881,4 +882,121 @@ public void
testSingleLedgerDirectoryCheckpointTriggerRemovePendingDeletedLedger
bookie.getLedgerStorage().flush();
Assert.assertEquals(pendingDeletedLedgers.size(), 0);
}
+
+ /**
+ * Simulates the full journal-missing scenario caused by concurrent
checkpointComplete
+ * calls in single-dir mode and verifies the monotonic fix prevents it.
+ *
+ * <p>Timeline of the original bug:
+ * <pre>
+ * 1. SDLS.flush() starts: newCheckpoint → mark(5, 0), begins flushing data
+ * 2. SyncThread starts: newCheckpoint → mark(7, 0), waits for flushMutex
+ * 3. SDLS.flush() completes flushing, releases flushMutex
+ * 4. SyncThread acquires flushMutex, finds writeCache empty, returns
+ * 5. SyncThread calls checkpointComplete(mark=7, compact=true) FIRST
+ * → rollLog(7), GC deletes journal files 3,4,5,6 (all with id < 7)
+ * 6. SDLS.flush() calls checkpointComplete(mark=5, compact=true) SECOND
+ * → rollLog(5) — OVERWRITES lastMark backwards from 7 to 5!
+ * → journal file 5 was already deleted in step 5
+ * 7. Bookie restarts: reads lastMark=5, looks for journal file 5 →
MISSING!
+ * → throws "Recovery log 5 is missing"
+ * </pre>
+ *
+ * <p>With the monotonic fix, step 6 is skipped (mark 5 <=
lastPersistedMark 7),
+ * so lastMark stays at 7 and the restart succeeds.
+ */
+ @Test
+ public void testConcurrentCheckpointCompleteJournalMissing() throws
Exception {
+ File baseDir = new File(tmpDir, "journalMissingTest");
+ File ledgerDir = new File(baseDir, "ledger");
+ File journalBaseDir = new File(baseDir, "journal");
+ ServerConfiguration conf =
TestBKConfiguration.newServerConfiguration();
+ conf.setGcWaitTime(1000);
+ conf.setProperty(DbLedgerStorage.WRITE_CACHE_MAX_SIZE_MB, 4);
+ conf.setProperty(DbLedgerStorage.READ_AHEAD_CACHE_MAX_SIZE_MB, 4);
+ conf.setLedgerStorageClass(DbLedgerStorage.class.getName());
+ conf.setLedgerDirNames(new String[] { ledgerDir.getCanonicalPath() });
+ conf.setJournalDirName(journalBaseDir.getCanonicalPath());
+ // Set maxBackupJournals to 0 so GC aggressively deletes all old
journals
+ conf.setMaxBackupJournals(0);
+
+ BookieImpl bookie = new TestBookieImpl(conf);
+ try {
+ Journal journal = bookie.getJournals().get(0);
+ File journalDir = journal.getJournalDirectory();
+ File ledgerDirMark = new File(ledgerDir + "/current", "lastMark");
+
+ // Create fake journal files: 3.txn, 4.txn, 5.txn, 6.txn, 7.txn,
8.txn
+ for (long id = 3; id <= 8; id++) {
+ File journalFile = new File(journalDir, Long.toHexString(id) +
".txn");
+ assertTrue("Failed to create journal file " + id,
journalFile.createNewFile());
+ }
+
+ // Verify all journal files exist
+ for (long id = 3; id <= 8; id++) {
+ File journalFile = new File(journalDir, Long.toHexString(id) +
".txn");
+ assertTrue("Journal file " + id + " should exist",
journalFile.exists());
+ }
+
+ CheckpointSource checkpointSource = new
CheckpointSourceList(bookie.getJournals());
+
+ // === Simulate the race ===
+
+ // Step 1: SDLS.flush() captures checkpoint at mark(5, 100)
+ journal.getLastLogMark().getCurMark().setLogMark(5, 100);
+ CheckpointSource.Checkpoint cpFlush =
checkpointSource.newCheckpoint();
+
+ // Step 2: SyncThread captures checkpoint at mark(7, 200) — newer
position
+ journal.getLastLogMark().getCurMark().setLogMark(7, 200);
+ CheckpointSource.Checkpoint cpSync =
checkpointSource.newCheckpoint();
+
+ // Step 5: SyncThread completes FIRST — checkpointComplete(mark=7,
compact=true)
+ // This should: rollLog to 7, GC journals with id < 7 (deletes
3,4,5,6)
+ checkpointSource.checkpointComplete(cpSync, true);
+
+ LogMark markAfterSync = readLogMark(ledgerDirMark);
+ assertEquals("lastMark should be at 7 after SyncThread", 7,
markAfterSync.getLogFileId());
+ assertEquals(200, markAfterSync.getLogFileOffset());
+
+ // Verify journals 3,4,5,6 were GC'd, 7,8 still exist
+ for (long id = 3; id <= 6; id++) {
+ File journalFile = new File(journalDir, Long.toHexString(id) +
".txn");
+ assertFalse("Journal " + id + " should have been GC'd",
journalFile.exists());
+ }
+ for (long id = 7; id <= 8; id++) {
+ File journalFile = new File(journalDir, Long.toHexString(id) +
".txn");
+ assertTrue("Journal " + id + " should still exist",
journalFile.exists());
+ }
+
+ // Step 6: SDLS.flush() completes SECOND —
checkpointComplete(mark=5, compact=true)
+ // WITHOUT FIX: rollLog would overwrite lastMark to 5, but journal
5 is already deleted!
+ // WITH FIX: mark 5 <= lastPersistedMark 7, so this is skipped
entirely.
+ checkpointSource.checkpointComplete(cpFlush, true);
+
Review Comment:
This test doesn’t actually exercise the concurrency ordering that caused the
bug: both checkpointComplete calls run sequentially in the same thread. With
the current implementation, a stale checkpoint that *entered*
checkpointComplete before a newer one (but completes after) can still regress
lastMark; adding a thread/latch-based test that forces that interleaving would
better validate the fix.
##########
bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java:
##########
@@ -881,4 +882,121 @@ public void
testSingleLedgerDirectoryCheckpointTriggerRemovePendingDeletedLedger
bookie.getLedgerStorage().flush();
Assert.assertEquals(pendingDeletedLedgers.size(), 0);
}
+
+ /**
+ * Simulates the full journal-missing scenario caused by concurrent
checkpointComplete
+ * calls in single-dir mode and verifies the monotonic fix prevents it.
+ *
+ * <p>Timeline of the original bug:
+ * <pre>
+ * 1. SDLS.flush() starts: newCheckpoint → mark(5, 0), begins flushing data
+ * 2. SyncThread starts: newCheckpoint → mark(7, 0), waits for flushMutex
+ * 3. SDLS.flush() completes flushing, releases flushMutex
+ * 4. SyncThread acquires flushMutex, finds writeCache empty, returns
+ * 5. SyncThread calls checkpointComplete(mark=7, compact=true) FIRST
+ * → rollLog(7), GC deletes journal files 3,4,5,6 (all with id < 7)
+ * 6. SDLS.flush() calls checkpointComplete(mark=5, compact=true) SECOND
+ * → rollLog(5) — OVERWRITES lastMark backwards from 7 to 5!
+ * → journal file 5 was already deleted in step 5
+ * 7. Bookie restarts: reads lastMark=5, looks for journal file 5 →
MISSING!
+ * → throws "Recovery log 5 is missing"
+ * </pre>
+ *
+ * <p>With the monotonic fix, step 6 is skipped (mark 5 <=
lastPersistedMark 7),
+ * so lastMark stays at 7 and the restart succeeds.
+ */
+ @Test
+ public void testConcurrentCheckpointCompleteJournalMissing() throws
Exception {
+ File baseDir = new File(tmpDir, "journalMissingTest");
+ File ledgerDir = new File(baseDir, "ledger");
+ File journalBaseDir = new File(baseDir, "journal");
+ ServerConfiguration conf =
TestBKConfiguration.newServerConfiguration();
+ conf.setGcWaitTime(1000);
+ conf.setProperty(DbLedgerStorage.WRITE_CACHE_MAX_SIZE_MB, 4);
+ conf.setProperty(DbLedgerStorage.READ_AHEAD_CACHE_MAX_SIZE_MB, 4);
+ conf.setLedgerStorageClass(DbLedgerStorage.class.getName());
+ conf.setLedgerDirNames(new String[] { ledgerDir.getCanonicalPath() });
+ conf.setJournalDirName(journalBaseDir.getCanonicalPath());
+ // Set maxBackupJournals to 0 so GC aggressively deletes all old
journals
+ conf.setMaxBackupJournals(0);
+
+ BookieImpl bookie = new TestBookieImpl(conf);
+ try {
+ Journal journal = bookie.getJournals().get(0);
+ File journalDir = journal.getJournalDirectory();
+ File ledgerDirMark = new File(ledgerDir + "/current", "lastMark");
+
+ // Create fake journal files: 3.txn, 4.txn, 5.txn, 6.txn, 7.txn,
8.txn
+ for (long id = 3; id <= 8; id++) {
+ File journalFile = new File(journalDir, Long.toHexString(id) +
".txn");
+ assertTrue("Failed to create journal file " + id,
journalFile.createNewFile());
+ }
+
+ // Verify all journal files exist
+ for (long id = 3; id <= 8; id++) {
+ File journalFile = new File(journalDir, Long.toHexString(id) +
".txn");
+ assertTrue("Journal file " + id + " should exist",
journalFile.exists());
+ }
+
+ CheckpointSource checkpointSource = new
CheckpointSourceList(bookie.getJournals());
+
+ // === Simulate the race ===
+
+ // Step 1: SDLS.flush() captures checkpoint at mark(5, 100)
+ journal.getLastLogMark().getCurMark().setLogMark(5, 100);
+ CheckpointSource.Checkpoint cpFlush =
checkpointSource.newCheckpoint();
+
+ // Step 2: SyncThread captures checkpoint at mark(7, 200) — newer
position
+ journal.getLastLogMark().getCurMark().setLogMark(7, 200);
+ CheckpointSource.Checkpoint cpSync =
checkpointSource.newCheckpoint();
+
+ // Step 5: SyncThread completes FIRST — checkpointComplete(mark=7,
compact=true)
+ // This should: rollLog to 7, GC journals with id < 7 (deletes
3,4,5,6)
+ checkpointSource.checkpointComplete(cpSync, true);
+
+ LogMark markAfterSync = readLogMark(ledgerDirMark);
+ assertEquals("lastMark should be at 7 after SyncThread", 7,
markAfterSync.getLogFileId());
+ assertEquals(200, markAfterSync.getLogFileOffset());
+
+ // Verify journals 3,4,5,6 were GC'd, 7,8 still exist
+ for (long id = 3; id <= 6; id++) {
+ File journalFile = new File(journalDir, Long.toHexString(id) +
".txn");
+ assertFalse("Journal " + id + " should have been GC'd",
journalFile.exists());
+ }
+ for (long id = 7; id <= 8; id++) {
+ File journalFile = new File(journalDir, Long.toHexString(id) +
".txn");
+ assertTrue("Journal " + id + " should still exist",
journalFile.exists());
+ }
+
+ // Step 6: SDLS.flush() completes SECOND —
checkpointComplete(mark=5, compact=true)
+ // WITHOUT FIX: rollLog would overwrite lastMark to 5, but journal
5 is already deleted!
+ // WITH FIX: mark 5 <= lastPersistedMark 7, so this is skipped
entirely.
+ checkpointSource.checkpointComplete(cpFlush, true);
+
+ // Verify: lastMark must NOT regress to 5. Should stay at 7.
+ LogMark markAfterFlush = readLogMark(ledgerDirMark);
+ assertEquals("lastMark must not regress after older checkpoint
completes",
+ 7, markAfterFlush.getLogFileId());
+ assertEquals(200, markAfterFlush.getLogFileOffset());
+
+ // Step 7: Simulate bookie restart — read lastMark and check
journal exists
+ journal.getLastLogMark().readLog();
+ LogMark restartMark = journal.getLastLogMark().getCurMark();
+ assertEquals("Reloaded lastMark should be 7", 7,
restartMark.getLogFileId());
Review Comment:
The “Simulate bookie restart” section calls lastLogMark.readLog() on the
existing Journal instance whose in-memory curMark has already been set to 7
earlier in the test. Since readLog() only updates curMark if the on-disk mark
is greater, this doesn’t validate that lastMark is actually being reloaded. To
better simulate a restart, reset curMark before readLog() or create a new
Bookie/Journal instance pointing at the same dirs and assert it loads the
persisted mark.
```suggestion
// Step 7: Simulate bookie restart — clear in-memory state, then
reload lastMark from disk
journal.getLastLogMark().getCurMark().setLogMark(0, 0);
journal.getLastLogMark().readLog();
LogMark restartMark = journal.getLastLogMark().getCurMark();
assertEquals("Reloaded lastMark should be 7", 7,
restartMark.getLogFileId());
assertEquals("Reloaded lastMark offset should be restored from
disk",
200, restartMark.getLogFileOffset());
```
##########
bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java:
##########
@@ -881,4 +882,121 @@ public void
testSingleLedgerDirectoryCheckpointTriggerRemovePendingDeletedLedger
bookie.getLedgerStorage().flush();
Assert.assertEquals(pendingDeletedLedgers.size(), 0);
}
+
+ /**
+ * Simulates the full journal-missing scenario caused by concurrent
checkpointComplete
+ * calls in single-dir mode and verifies the monotonic fix prevents it.
+ *
+ * <p>Timeline of the original bug:
+ * <pre>
+ * 1. SDLS.flush() starts: newCheckpoint → mark(5, 0), begins flushing data
+ * 2. SyncThread starts: newCheckpoint → mark(7, 0), waits for flushMutex
+ * 3. SDLS.flush() completes flushing, releases flushMutex
+ * 4. SyncThread acquires flushMutex, finds writeCache empty, returns
+ * 5. SyncThread calls checkpointComplete(mark=7, compact=true) FIRST
+ * → rollLog(7), GC deletes journal files 3,4,5,6 (all with id < 7)
+ * 6. SDLS.flush() calls checkpointComplete(mark=5, compact=true) SECOND
+ * → rollLog(5) — OVERWRITES lastMark backwards from 7 to 5!
+ * → journal file 5 was already deleted in step 5
+ * 7. Bookie restarts: reads lastMark=5, looks for journal file 5 →
MISSING!
+ * → throws "Recovery log 5 is missing"
+ * </pre>
+ *
+ * <p>With the monotonic fix, step 6 is skipped (mark 5 <=
lastPersistedMark 7),
+ * so lastMark stays at 7 and the restart succeeds.
+ */
+ @Test
+ public void testConcurrentCheckpointCompleteJournalMissing() throws
Exception {
+ File baseDir = new File(tmpDir, "journalMissingTest");
+ File ledgerDir = new File(baseDir, "ledger");
+ File journalBaseDir = new File(baseDir, "journal");
+ ServerConfiguration conf =
TestBKConfiguration.newServerConfiguration();
+ conf.setGcWaitTime(1000);
+ conf.setProperty(DbLedgerStorage.WRITE_CACHE_MAX_SIZE_MB, 4);
+ conf.setProperty(DbLedgerStorage.READ_AHEAD_CACHE_MAX_SIZE_MB, 4);
+ conf.setLedgerStorageClass(DbLedgerStorage.class.getName());
+ conf.setLedgerDirNames(new String[] { ledgerDir.getCanonicalPath() });
+ conf.setJournalDirName(journalBaseDir.getCanonicalPath());
+ // Set maxBackupJournals to 0 so GC aggressively deletes all old
journals
+ conf.setMaxBackupJournals(0);
+
+ BookieImpl bookie = new TestBookieImpl(conf);
+ try {
+ Journal journal = bookie.getJournals().get(0);
+ File journalDir = journal.getJournalDirectory();
+ File ledgerDirMark = new File(ledgerDir + "/current", "lastMark");
+
+ // Create fake journal files: 3.txn, 4.txn, 5.txn, 6.txn, 7.txn,
8.txn
+ for (long id = 3; id <= 8; id++) {
+ File journalFile = new File(journalDir, Long.toHexString(id) +
".txn");
+ assertTrue("Failed to create journal file " + id,
journalFile.createNewFile());
+ }
+
+ // Verify all journal files exist
+ for (long id = 3; id <= 8; id++) {
+ File journalFile = new File(journalDir, Long.toHexString(id) +
".txn");
+ assertTrue("Journal file " + id + " should exist",
journalFile.exists());
+ }
+
+ CheckpointSource checkpointSource = new
CheckpointSourceList(bookie.getJournals());
+
+ // === Simulate the race ===
+
+ // Step 1: SDLS.flush() captures checkpoint at mark(5, 100)
+ journal.getLastLogMark().getCurMark().setLogMark(5, 100);
+ CheckpointSource.Checkpoint cpFlush =
checkpointSource.newCheckpoint();
+
+ // Step 2: SyncThread captures checkpoint at mark(7, 200) — newer
position
+ journal.getLastLogMark().getCurMark().setLogMark(7, 200);
+ CheckpointSource.Checkpoint cpSync =
checkpointSource.newCheckpoint();
+
+ // Step 5: SyncThread completes FIRST — checkpointComplete(mark=7,
compact=true)
+ // This should: rollLog to 7, GC journals with id < 7 (deletes
3,4,5,6)
+ checkpointSource.checkpointComplete(cpSync, true);
+
+ LogMark markAfterSync = readLogMark(ledgerDirMark);
+ assertEquals("lastMark should be at 7 after SyncThread", 7,
markAfterSync.getLogFileId());
+ assertEquals(200, markAfterSync.getLogFileOffset());
+
+ // Verify journals 3,4,5,6 were GC'd, 7,8 still exist
+ for (long id = 3; id <= 6; id++) {
+ File journalFile = new File(journalDir, Long.toHexString(id) +
".txn");
+ assertFalse("Journal " + id + " should have been GC'd",
journalFile.exists());
+ }
+ for (long id = 7; id <= 8; id++) {
+ File journalFile = new File(journalDir, Long.toHexString(id) +
".txn");
+ assertTrue("Journal " + id + " should still exist",
journalFile.exists());
+ }
+
+ // Step 6: SDLS.flush() completes SECOND —
checkpointComplete(mark=5, compact=true)
+ // WITHOUT FIX: rollLog would overwrite lastMark to 5, but journal
5 is already deleted!
+ // WITH FIX: mark 5 <= lastPersistedMark 7, so this is skipped
entirely.
+ checkpointSource.checkpointComplete(cpFlush, true);
+
+ // Verify: lastMark must NOT regress to 5. Should stay at 7.
+ LogMark markAfterFlush = readLogMark(ledgerDirMark);
+ assertEquals("lastMark must not regress after older checkpoint
completes",
+ 7, markAfterFlush.getLogFileId());
+ assertEquals(200, markAfterFlush.getLogFileOffset());
+
+ // Step 7: Simulate bookie restart — read lastMark and check
journal exists
+ journal.getLastLogMark().readLog();
+ LogMark restartMark = journal.getLastLogMark().getCurMark();
+ assertEquals("Reloaded lastMark should be 7", 7,
restartMark.getLogFileId());
+
+ // The journal file pointed to by lastMark must exist
+ File markedJournal = new File(journalDir,
+ Long.toHexString(restartMark.getLogFileId()) + ".txn");
+ assertTrue("Journal file " + restartMark.getLogFileId() + " must
exist for recovery",
+ markedJournal.exists());
+
+ // Verify that listJournalIds finds the expected journal for replay
+ List<Long> replayLogs = Journal.listJournalIds(journalDir,
+ journalId -> journalId >= restartMark.getLogFileId());
+ assertTrue("Replay journal list must contain the marked journal",
+ replayLogs.size() > 0 && replayLogs.get(0) ==
restartMark.getLogFileId());
+ } finally {
+ bookie.getLedgerStorage().shutdown();
Review Comment:
Cleanup only shuts down the LedgerStorage. BookieImpl also starts SyncThread
and Journal threads; leaving them running can cause thread leaks and test
flakiness. Prefer calling bookie.shutdown() (or otherwise stopping
journals/sync thread) in the finally block.
```suggestion
bookie.shutdown();
```
--
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]
