Fix DbLedgerStorage checkpoint losing ledger metadata when write cache is empty

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java

@@ -807,6 +807,10 @@ public void checkpoint(Checkpoint checkpoint) throws IOException {
 
         try {
             if (writeCache.isEmpty()) {
+                // Ledger metadata updates can be journaled without any pending entry data.
+                // Persist them before allowing the journal checkpoint mark to advance.
+                flushLedgerIndex();
+                lastCheckpoint = thisCheckpoint;
                 return;
             }
             // Swap the write cache so that writes can continue to happen while the flush is
@@ -840,9 +844,7 @@ public void checkpoint(Checkpoint checkpoint) throws IOException {
                         .log("DB batch flushed");
             }
 
-            long ledgerIndexStartTime = MathUtils.nowInNano();
-            ledgerIndex.flush();
-            recordSuccessfulEvent(dbLedgerStorageStats.getFlushLedgerIndexStats(), ledgerIndexStartTime);
+            flushLedgerIndex();
 
             lastCheckpoint = thisCheckpoint;
 
@@ -885,6 +887,12 @@ public void checkpoint(Checkpoint checkpoint) throws IOException {
         }
     }
 
+    private void flushLedgerIndex() throws IOException {
+        long ledgerIndexStartTime = MathUtils.nowInNano();
+        ledgerIndex.flush();
+        recordSuccessfulEvent(dbLedgerStorageStats.getFlushLedgerIndexStats(), ledgerIndexStartTime);
+    }
+
     /**
      * Swap the current write cache with the replacement cache.
      */

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

@@ -108,6 +108,21 @@ public void teardown() throws Exception {
         tmpDir.delete();
     }
 
+    private void addEntryAndFlush(long ledgerId, long entryId) throws Exception {
+        storage.setMasterKey(ledgerId, "key".getBytes());
+
+        ByteBuf entry = Unpooled.buffer(1024);
+        entry.writeLong(ledgerId);
+        entry.writeLong(entryId);
+        entry.writeLong(entryId - 1);
+        entry.writeBytes(("entry-" + entryId).getBytes());
+
+        storage.addEntry(entry);
+        storage.flush();
+
+        assertFalse(storage.isFlushRequired());
+    }
+
     @Test
     public void simple() throws Exception {
         assertEquals(false, storage.ledgerExists(3));
@@ -232,6 +247,57 @@ public void simple() throws Exception {
         }
     }
 
+    @Test
+    public void testFlushPersistsFencedMetadataWithoutPendingEntries() throws Exception {
+        long ledgerId = 1;
+        addEntryAndFlush(ledgerId, 0);
+
+        assertTrue(storage.setFenced(ledgerId));
+        assertFalse(storage.isFlushRequired());
+
+        storage.flush();
+        storage.shutdown();
+
+        Bookie restartedBookie = new TestBookieImpl(conf);
+        DbLedgerStorage restartedStorage = (DbLedgerStorage) restartedBookie.getLedgerStorage();
+        try {
+            assertTrue(restartedStorage.isFenced(ledgerId));
+        } finally {
+            restartedStorage.shutdown();
+        }
+
+        storage = (DbLedgerStorage) new TestBookieImpl(conf).getLedgerStorage();
+    }
+
+    @Test
+    public void testFlushPersistsExplicitLacMetadataWithoutPendingEntries() throws Exception {
+        long ledgerId = 1;
+        addEntryAndFlush(ledgerId, 0);
+
+        ByteBuf explicitLac = Unpooled.buffer(Long.BYTES * 2);
+        explicitLac.writeLong(ledgerId);
+        explicitLac.writeLong(0);
+        storage.setExplicitLac(ledgerId, explicitLac);
+        assertFalse(storage.isFlushRequired());
+
+        storage.flush();
+        storage.shutdown();
+
+        Bookie restartedBookie = new TestBookieImpl(conf);
+        DbLedgerStorage restartedStorage = (DbLedgerStorage) restartedBookie.getLedgerStorage();
+        ByteBuf recoveredExplicitLac = null;
+        try {
+            recoveredExplicitLac = restartedStorage.getExplicitLac(ledgerId);
+            Assert.assertNotNull(recoveredExplicitLac);
+            assertEquals(0, ByteBufUtil.compare(explicitLac, recoveredExplicitLac));
+        } finally {
+            ReferenceCountUtil.release(recoveredExplicitLac);
+            restartedStorage.shutdown();
+        }
+
+        storage = (DbLedgerStorage) new TestBookieImpl(conf).getLedgerStorage();
+    }
+
     @Test
     public void testBookieCompaction() throws Exception {
         storage.setMasterKey(4, "key".getBytes());