This is an automated email from the ASF dual-hosted git repository. hangc0276 pushed a commit to branch branch-4.18 in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
commit 96230ec67654ea876edac724d33d4c19c02d216f Author: void-ptr974 <[email protected]> AuthorDate: Thu May 28 09:27:39 2026 +0800 fix Bookkeeper can't startup cause by 'IOException: Recovery log xxx is missing' (#4740) * fix #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 #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 #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 #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 (cherry picked from commit ccfabfaf50efccdb3f8cec46e6243a76cac04ad7) --- .../java/org/apache/bookkeeper/bookie/Journal.java | 30 ++- .../bookie/storage/ldb/DbLedgerStorageTest.java | 240 +++++++++++++++++++++ 2 files changed, 269 insertions(+), 1 deletion(-) diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java index ae0ded3721..830763aa02 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java @@ -628,6 +628,10 @@ public class Journal implements CheckpointSource { 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 class Journal implements CheckpointSource { 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 class Journal implements CheckpointSource { /** * 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 class Journal implements CheckpointSource { 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 class Journal implements CheckpointSource { } } + @VisibleForTesting + protected void persistLastLogMark(LastLogMark mark) throws NoWritableLedgerDirException { + mark.rollLog(mark); + } + + @VisibleForTesting + protected boolean isCheckpointCompleteLockHeldByCurrentThread() { + return Thread.holdsLock(checkpointLock); + } + /** * Scan the journal. * diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java index dfc2459678..a569a638cc 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java @@ -32,11 +32,17 @@ import io.netty.buffer.Unpooled; 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,7 +51,9 @@ import org.apache.bookkeeper.bookie.CheckpointSource; 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; @@ -53,6 +61,7 @@ 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 class DbLedgerStorageTest { 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(); + } + } + } }
