fix Bookkeeper can't startup cause by 'IOException: Recovery log xxx is missing' (apache#4740)

* fix apache#4105

When journal and dbledgerstorage on the same disk, concurrent checkpointComplete calls from SyncThread and SingleDirectoryDbLedgerStorage.flush() can overwrite lastMark backwards,causing journal files to be deleted while still referenced by the older mark.

* fix apache#4105

When singleLedgerDirs=true, bookie can crash on restart with "Recovery log is missing" due to lastMark file being overwritten backwards.

  Root cause

  SyncThread.flush() has a nested call pattern that causes lastMark regression:

  SyncThread.flush():
    1. outerCheckpoint = newCheckpoint()        → mark = (file 5, offset 100)
    2. ledgerStorage.flush()
       → SingleDirectoryDbLedgerStorage.flush():
         a. innerCheckpoint = newCheckpoint()   → mark = (file 7, offset 200), journal has advanced
         b. flushes data to disk
         c. checkpointComplete(mark=7, compact=true)
            → persists lastMark=7, deletes journal files with id < 7 (including file 5)
    3. checkpointComplete(mark=5, compact=false)
       → persists lastMark=5, overwriting the 7 written in step 2c
    4. Bookie restarts → reads lastMark=5 → file 5 no longer exists → crash

  Step 1 captures the checkpoint before step 2 runs, so it is always older. Step 2c advances lastMark and garbage-collects old journals. Step 3 then overwrites lastMark backwards to a
  position whose journal file was already deleted.

  Conditions

  - singleLedgerDirs=true (journal and ledger on the same disk, so SingleDirectoryDbLedgerStorage.flush() calls checkpointComplete internally)
  - Journal file rotates between the two newCheckpoint() calls (requires sufficient write throughput)
  - maxBackupJournals small enough for old files to actually be deleted

  Fix

  Add a monotonic guard in checkpointComplete(): track the highest mark persisted so far, skip any call with an older mark. This prevents rollLog from overwriting lastMark backwards.

* fix apache#4105

When singleLedgerDirs=true, bookie can crash on restart with "Recovery log is missing" due to lastMark file being overwritten backwards.

  Root cause

  SyncThread.flush() has a nested call pattern that causes lastMark regression:

  SyncThread.flush():
    1. outerCheckpoint = newCheckpoint()        → mark = (file 5, offset 100)
    2. ledgerStorage.flush()
       → SingleDirectoryDbLedgerStorage.flush():
         a. innerCheckpoint = newCheckpoint()   → mark = (file 7, offset 200), journal has advanced
         b. flushes data to disk
         c. checkpointComplete(mark=7, compact=true)
            → persists lastMark=7, deletes journal files with id < 7 (including file 5)
    3. checkpointComplete(mark=5, compact=false)
       → persists lastMark=5, overwriting the 7 written in step 2c
    4. Bookie restarts → reads lastMark=5 → file 5 no longer exists → crash

  Step 1 captures the checkpoint before step 2 runs, so it is always older. Step 2c advances lastMark and garbage-collects old journals. Step 3 then overwrites lastMark backwards to a
  position whose journal file was already deleted.

  Conditions

  - singleLedgerDirs=true (journal and ledger on the same disk, so SingleDirectoryDbLedgerStorage.flush() calls checkpointComplete internally)
  - Journal file rotates between the two newCheckpoint() calls (requires sufficient write throughput)
  - maxBackupJournals small enough for old files to actually be deleted

  Fix

  Add a monotonic guard in checkpointComplete(): track the highest mark persisted so far, skip any call with an older mark. This prevents rollLog from overwriting lastMark backwards.

* fix apache#4105

When singleLedgerDirs=true, bookie can crash on restart with "Recovery log is missing" due to lastMark file being overwritten backwards.

  Root cause

  SyncThread.flush() has a nested call pattern that causes lastMark regression:

  SyncThread.flush():
    1. outerCheckpoint = newCheckpoint()        → mark = (file 5, offset 100)
    2. ledgerStorage.flush()
       → SingleDirectoryDbLedgerStorage.flush():
         a. innerCheckpoint = newCheckpoint()   → mark = (file 7, offset 200), journal has advanced
         b. flushes data to disk
         c. checkpointComplete(mark=7, compact=true)
            → persists lastMark=7, deletes journal files with id < 7 (including file 5)
    3. checkpointComplete(mark=5, compact=false)
       → persists lastMark=5, overwriting the 7 written in step 2c
    4. Bookie restarts → reads lastMark=5 → file 5 no longer exists → crash

  Step 1 captures the checkpoint before step 2 runs, so it is always older. Step 2c advances lastMark and garbage-collects old journals. Step 3 then overwrites lastMark backwards to a
  position whose journal file was already deleted.

  Conditions

  - singleLedgerDirs=true (journal and ledger on the same disk, so SingleDirectoryDbLedgerStorage.flush() calls checkpointComplete internally)
  - Journal file rotates between the two newCheckpoint() calls (requires sufficient write throughput)
  - maxBackupJournals small enough for old files to actually be deleted

  Fix

  Add a monotonic guard in checkpointComplete(): track the highest mark persisted so far, skip any call with an older mark. This prevents rollLog from overwriting lastMark backwards.

* Fix lastMark regression in checkpoint completion

Author: void-ptr974

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java

@@ -628,6 +628,10 @@ static void writePaddingBytes(JournalChannel jc, ByteBuf paddingBuffer, int jour
 
     private final LastLogMark lastLogMark = new LastLogMark(0, 0);
 
+    // Ensures lastMark only advances forward across concurrent checkpointComplete calls.
+    private final Object checkpointLock = new Object();
+    private final LogMark lastPersistedMark = new LogMark(0, 0);
+
     private static final String LAST_MARK_DEFAULT_NAME = "lastMark";
 
     private final String lastMarkFileName;
@@ -705,6 +709,8 @@ public Journal(int journalIndex, File journalDirectory, ServerConfiguration conf
             lastMarkFileName = LAST_MARK_DEFAULT_NAME + "." + journalIndex;
         }
         lastLogMark.readLog();
+        lastPersistedMark.setLogMark(
+                lastLogMark.getCurMark().getLogFileId(), lastLogMark.getCurMark().getLogFileOffset());
         log.debug().attr("lastMark", () -> lastLogMark.getCurMark()).log("Last Log Mark");
 
         try {
@@ -771,6 +777,9 @@ public Checkpoint newCheckpoint() {
     /**
      * Telling journal a checkpoint is finished.
      *
+     * <p>Skips if the checkpoint is older than the last persisted mark,
+     * preventing lastMark from regressing backwards.
+     *
      * @throws IOException
      */
     @Override
@@ -781,7 +790,16 @@ public void checkpointComplete(Checkpoint checkpoint, boolean compact) throws IO
         LogMarkCheckpoint lmcheckpoint = (LogMarkCheckpoint) checkpoint;
         LastLogMark mark = lmcheckpoint.mark;
 
-        mark.rollLog(mark);
+        // See #4105: keep the monotonic decision and lastMark persistence together.
+        synchronized (checkpointLock) {
+            if (mark.getCurMark().compare(lastPersistedMark) < 0) {
+                return;
+            }
+            persistLastLogMark(mark);
+            lastPersistedMark.setLogMark(
+                    mark.getCurMark().getLogFileId(), mark.getCurMark().getLogFileOffset());
+        }
+
         if (compact) {
             // list the journals that have been marked
             List<Long> logs = listJournalIds(journalDirectory, new JournalRollingFilter(mark));
@@ -803,6 +821,16 @@ public void checkpointComplete(Checkpoint checkpoint, boolean compact) throws IO
         }
     }
 
+    @VisibleForTesting
+    protected void persistLastLogMark(LastLogMark mark) throws NoWritableLedgerDirException {
+        mark.rollLog(mark);
+    }
+
+    @VisibleForTesting
+    protected boolean isCheckpointCompleteLockHeldByCurrentThread() {
+        return Thread.holdsLock(checkpointLock);
+    }
+
     /**
      * Scan the journal.
      *

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java

@@ -32,11 +32,17 @@
 import io.netty.util.ReferenceCountUtil;
 import java.io.File;
 import java.io.FileInputStream;
+import java.io.FileOutputStream;
 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import org.apache.bookkeeper.bookie.Bookie;
 import org.apache.bookkeeper.bookie.Bookie.NoEntryException;
 import org.apache.bookkeeper.bookie.BookieException;
@@ -45,14 +51,17 @@
 import org.apache.bookkeeper.bookie.CheckpointSourceList;
 import org.apache.bookkeeper.bookie.DefaultEntryLogger;
 import org.apache.bookkeeper.bookie.EntryLocation;
+import org.apache.bookkeeper.bookie.Journal;
 import org.apache.bookkeeper.bookie.LedgerDirsManager;
+import org.apache.bookkeeper.bookie.LedgerDirsManager.NoWritableLedgerDirException;
 import org.apache.bookkeeper.bookie.LedgerStorage;
 import org.apache.bookkeeper.bookie.LogMark;
 import org.apache.bookkeeper.bookie.TestBookieImpl;
 import org.apache.bookkeeper.bookie.storage.EntryLogger;
 import org.apache.bookkeeper.conf.ServerConfiguration;
 import org.apache.bookkeeper.conf.TestBKConfiguration;
 import org.apache.bookkeeper.proto.BookieProtocol;
+import org.apache.bookkeeper.util.DiskChecker;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -881,4 +890,235 @@ public void testSingleLedgerDirectoryCheckpointTriggerRemovePendingDeletedLedger
         bookie.getLedgerStorage().flush();
         Assert.assertEquals(pendingDeletedLedgers.size(), 0);
     }
+
+    @Test
+    public void testCheckpointCompleteSkipsStaleCheckpointAfterNewerCheckpoint() throws Exception {
+        File baseDir = new File(tmpDir, "journalStaleCheckpointTest");
+        File ledgerDir = new File(baseDir, "ledger");
+        File journalBaseDir = new File(baseDir, "journal");
+        File journalDir = BookieImpl.getCurrentDirectory(journalBaseDir);
+        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());
+        conf.setMaxBackupJournals(0);
+
+        BookieImpl.checkDirectoryStructure(journalDir);
+        LedgerDirsManager dirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
+                new DiskChecker(0.95f, 0.90f));
+        Journal journal = new Journal(0, journalDir, conf, dirsManager);
+        File ledgerDirMark = new File(BookieImpl.getCurrentDirectory(ledgerDir), "lastMark");
+        createJournalFiles(journalDir, 3, 10);
+
+        CheckpointSource checkpointSource = new CheckpointSourceList(Lists.newArrayList(journal));
+
+        journal.getLastLogMark().getCurMark().setLogMark(7, 100);
+        CheckpointSource.Checkpoint staleCheckpoint = checkpointSource.newCheckpoint();
+
+        journal.getLastLogMark().getCurMark().setLogMark(9, 300);
+        CheckpointSource.Checkpoint newerCheckpoint = checkpointSource.newCheckpoint();
+
+        checkpointSource.checkpointComplete(newerCheckpoint, true);
+        checkpointSource.checkpointComplete(staleCheckpoint, true);
+
+        LogMark markAfterStaleCheckpoint = readLogMark(ledgerDirMark);
+        assertEquals("lastMark must stay at newer checkpoint",
+                9, markAfterStaleCheckpoint.getLogFileId());
+        assertEquals(300, markAfterStaleCheckpoint.getLogFileOffset());
+
+        for (long id = 3; id <= 8; id++) {
+            File journalFile = new File(journalDir, Long.toHexString(id) + ".txn");
+            assertFalse("Journal " + id + " should have been garbage collected", journalFile.exists());
+        }
+        for (long id = 9; id <= 10; id++) {
+            File journalFile = new File(journalDir, Long.toHexString(id) + ".txn");
+            assertTrue("Journal " + id + " should still exist", journalFile.exists());
+        }
+    }
+
+    @Test
+    public void testCheckpointCompleteUsesReloadedLastMarkAsPersistedMark() throws Exception {
+        File baseDir = new File(tmpDir, "journalReloadedLastMarkTest");
+        File ledgerDir = new File(baseDir, "ledger");
+        File journalBaseDir = new File(baseDir, "journal");
+        File journalDir = BookieImpl.getCurrentDirectory(journalBaseDir);
+        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());
+        conf.setMaxBackupJournals(0);
+
+        BookieImpl.checkDirectoryStructure(journalDir);
+        LedgerDirsManager dirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
+                new DiskChecker(0.95f, 0.90f));
+        File ledgerDirMark = new File(BookieImpl.getCurrentDirectory(ledgerDir), "lastMark");
+        writeLogMark(ledgerDirMark, 9, 300);
+        createJournalFiles(journalDir, 7, 10);
+
+        Journal journal = new Journal(0, journalDir, conf, dirsManager);
+        CheckpointSource checkpointSource = new CheckpointSourceList(Lists.newArrayList(journal));
+
+        journal.getLastLogMark().getCurMark().setLogMark(7, 100);
+        CheckpointSource.Checkpoint staleCheckpoint = checkpointSource.newCheckpoint();
+        checkpointSource.checkpointComplete(staleCheckpoint, true);
+
+        LogMark markAfterStaleCheckpoint = readLogMark(ledgerDirMark);
+        assertEquals("Reloaded lastMark must seed the monotonic guard",
+                9, markAfterStaleCheckpoint.getLogFileId());
+        assertEquals(300, markAfterStaleCheckpoint.getLogFileOffset());
+    }
+
+    @Test
+    public void testConcurrentCheckpointCompleteLastMarkRegression() throws Exception {
+        File baseDir = new File(tmpDir, "journalMissingTest");
+        File ledgerDir = new File(baseDir, "ledger");
+        File journalBaseDir = new File(baseDir, "journal");
+        File journalDir = BookieImpl.getCurrentDirectory(journalBaseDir);
+        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());
+        conf.setMaxBackupJournals(0);
+
+        BookieImpl.checkDirectoryStructure(journalDir);
+        LedgerDirsManager dirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
+                new DiskChecker(0.95f, 0.90f));
+        CountDownLatch stalePersistStarted = new CountDownLatch(1);
+        CountDownLatch releaseStalePersist = new CountDownLatch(1);
+        CountDownLatch newerPersisted = new CountDownLatch(1);
+        BlockingJournal journal = new BlockingJournal(0, journalDir, conf, dirsManager,
+                stalePersistStarted, releaseStalePersist, newerPersisted);
+        File ledgerDirMark = new File(BookieImpl.getCurrentDirectory(ledgerDir), "lastMark");
+        createJournalFiles(journalDir, 3, 10);
+
+        CheckpointSource checkpointSource = new CheckpointSourceList(Lists.newArrayList(journal));
+
+        journal.getLastLogMark().getCurMark().setLogMark(7, 100);
+        CheckpointSource.Checkpoint staleCheckpoint = checkpointSource.newCheckpoint();
+
+        journal.getLastLogMark().getCurMark().setLogMark(9, 300);
+        CheckpointSource.Checkpoint newerCheckpoint = checkpointSource.newCheckpoint();
+
+        ExecutorService executor = Executors.newFixedThreadPool(2);
+        try {
+            Future<?> staleFuture = executor.submit(() -> {
+                checkpointSource.checkpointComplete(staleCheckpoint, true);
+                return null;
+            });
+            assertTrue("Stale checkpoint should reach lastMark persistence",
+                    stalePersistStarted.await(10, TimeUnit.SECONDS));
+
+            Future<?> newerFuture = executor.submit(() -> {
+                checkpointSource.checkpointComplete(newerCheckpoint, true);
+                return null;
+            });
+            releaseStalePersist.countDown();
+
+            newerFuture.get(10, TimeUnit.SECONDS);
+            staleFuture.get(10, TimeUnit.SECONDS);
+        } finally {
+            executor.shutdownNow();
+        }
+
+        LogMark markAfterFlush = readLogMark(ledgerDirMark);
+        assertEquals("lastMark must not regress after concurrent checkpoints",
+                9, markAfterFlush.getLogFileId());
+        assertEquals(300, markAfterFlush.getLogFileOffset());
+
+        for (long id = 3; id <= 8; id++) {
+            File journalFile = new File(journalDir, Long.toHexString(id) + ".txn");
+            assertFalse("Journal " + id + " should have been garbage collected", journalFile.exists());
+        }
+        for (long id = 9; id <= 10; id++) {
+            File journalFile = new File(journalDir, Long.toHexString(id) + ".txn");
+            assertTrue("Journal " + id + " should still exist", journalFile.exists());
+        }
+
+        journal.getLastLogMark().getCurMark().setLogMark(0, 0);
+        journal.getLastLogMark().readLog();
+        LogMark restartMark = journal.getLastLogMark().getCurMark();
+        assertEquals("Reloaded lastMark should be 9", 9, restartMark.getLogFileId());
+        assertEquals("Reloaded lastMark offset should be restored from disk",
+                300, restartMark.getLogFileOffset());
+
+        File markedJournal = new File(journalDir,
+                Long.toHexString(restartMark.getLogFileId()) + ".txn");
+        assertTrue("Journal file " + restartMark.getLogFileId() + " must exist for recovery",
+                markedJournal.exists());
+
+        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());
+    }
+
+    private static void createJournalFiles(File journalDir, long startId, long endId) throws IOException {
+        for (long id = startId; id <= endId; id++) {
+            File journalFile = new File(journalDir, Long.toHexString(id) + ".txn");
+            assertTrue("Failed to create journal file " + id, journalFile.createNewFile());
+        }
+    }
+
+    private static void writeLogMark(File file, long logFileId, long logFileOffset) throws IOException {
+        byte[] buff = new byte[16];
+        ByteBuffer bb = ByteBuffer.wrap(buff);
+        LogMark mark = new LogMark(logFileId, logFileOffset);
+        mark.writeLogMark(bb);
+        try (FileOutputStream fos = new FileOutputStream(file)) {
+            fos.write(buff);
+            fos.getChannel().force(true);
+        }
+    }
+
+    private static void awaitLatch(CountDownLatch latch, String message) {
+        try {
+            assertTrue(message, latch.await(10, TimeUnit.SECONDS));
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new AssertionError(message, e);
+        }
+    }
+
+    private static final class BlockingJournal extends Journal {
+        private static final long STALE_LOG_ID = 7;
+        private static final long NEWER_LOG_ID = 9;
+
+        private final CountDownLatch stalePersistStarted;
+        private final CountDownLatch releaseStalePersist;
+        private final CountDownLatch newerPersisted;
+
+        BlockingJournal(int journalIndex, File journalDirectory, ServerConfiguration conf,
+                        LedgerDirsManager ledgerDirsManager, CountDownLatch stalePersistStarted,
+                        CountDownLatch releaseStalePersist, CountDownLatch newerPersisted) {
+            super(journalIndex, journalDirectory, conf, ledgerDirsManager);
+            this.stalePersistStarted = stalePersistStarted;
+            this.releaseStalePersist = releaseStalePersist;
+            this.newerPersisted = newerPersisted;
+        }
+
+        @Override
+        protected void persistLastLogMark(LastLogMark mark) throws NoWritableLedgerDirException {
+            long logFileId = mark.getCurMark().getLogFileId();
+            if (logFileId == STALE_LOG_ID) {
+                stalePersistStarted.countDown();
+                awaitLatch(releaseStalePersist, "Timed out waiting to release stale checkpoint");
+                if (!isCheckpointCompleteLockHeldByCurrentThread()) {
+                    awaitLatch(newerPersisted, "Timed out waiting for newer checkpoint to persist");
+                }
+            }
+            super.persistLastLogMark(mark);
+            if (logFileId == NEWER_LOG_ID) {
+                newerPersisted.countDown();
+            }
+        }
+    }
 }