Fix lastMark regression in checkpoint completion

Author: void-ptr974

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

@@ -704,6 +704,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());
         if (LOG.isDebugEnabled()) {
             LOG.debug("Last Log Mark : {}", lastLogMark.getCurMark());
         }
@@ -770,7 +772,7 @@ public Checkpoint newCheckpoint() {
     /**
      * Telling journal a checkpoint is finished.
      *
-     * <p>Skips if the checkpoint is not newer than the last persisted mark,
+     * <p>Skips if the checkpoint is older than the last persisted mark,
      * preventing lastMark from regressing backwards.
      *
      * @throws IOException
@@ -783,16 +785,16 @@ public void checkpointComplete(Checkpoint checkpoint, boolean compact) throws IO
         LogMarkCheckpoint lmcheckpoint = (LogMarkCheckpoint) checkpoint;
         LastLogMark mark = lmcheckpoint.mark;
 
-        // Skip if this mark is not newer than what was already persisted.
+        // 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());
         }
 
-        mark.rollLog(mark);
         if (compact) {
             // list the journals that have been marked
             List<Long> logs = listJournalIds(journalDirectory, new JournalRollingFilter(mark));
@@ -814,6 +816,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.
      *