Fix the data loss issue that caused by the wrong entry log header

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

@@ -1151,6 +1151,10 @@ EntryLogMetadata extractEntryLogMetadataFromIndex(long entryLogId) throws IOExce
                     + " -- found: " + meta.getLedgersMap().size() + " -- entryLogId: " + entryLogId);
         }
 
+        if (header.ledgersCount == 0) {
+            throw new IOException("No ledgers map found in entryLogId " + entryLogId + ", do scan to double confirm");
+        }
+
         return meta;
     }
 

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

@@ -85,6 +85,7 @@ class EntryLoggerAllocator {
         // within the same JVM. All of these Bookie instances access this header
         // so there can be race conditions when entry logs are rolled over and
         // this header buffer is cleared before writing it into the new logChannel.
+        logfileHeader.setZero(0, DefaultEntryLogger.LOGFILE_HEADER_SIZE);
         logfileHeader.writeBytes("BKLO".getBytes(UTF_8));
         logfileHeader.writeInt(DefaultEntryLogger.HEADER_CURRENT_VERSION);
         logfileHeader.writerIndex(DefaultEntryLogger.LOGFILE_HEADER_SIZE);

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

@@ -842,11 +842,12 @@ protected void extractMetaFromEntryLogs() throws EntryLogMetadataMapException {
                 continue;
             }
 
-            LOG.info("Extracting entry log meta from entryLogId: {}", entryLogId);
 
             try {
                 // Read through the entry log file and extract the entry log meta
                 EntryLogMetadata entryLogMeta = entryLogger.getEntryLogMetadata(entryLogId, throttler);
+                LOG.info("Extracted entry log meta from entryLogId: {}, ledgers {}",
+                    entryLogId, entryLogMeta.getLedgersMap().keys());
                 removeIfLedgerNotExists(entryLogMeta);
                 if (entryLogMeta.isEmpty()) {
                     // This means the entry log is not associated with any active

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/DefaultEntryLogTest.java

@@ -391,6 +391,46 @@ public void testRecoverFromLedgersMap() throws Exception {
         assertEquals(120, meta.getRemainingSize());
     }
 
+    @Test
+    public void testLedgersMapIsEmpty() throws Exception {
+        // create some entries
+        entryLogger.addEntry(1L, generateEntry(1, 1).nioBuffer());
+        entryLogger.addEntry(3L, generateEntry(3, 1).nioBuffer());
+        entryLogger.addEntry(2L, generateEntry(2, 1).nioBuffer());
+        entryLogger.addEntry(1L, generateEntry(1, 2).nioBuffer());
+        ((EntryLogManagerBase) entryLogger.getEntryLogManager()).createNewLog(DefaultEntryLogger.UNASSIGNED_LEDGERID);
+        entryLogger.close();
+
+        // Rewrite the entry log header to be on V0 format
+        File f = new File(curDir, "0.log");
+        RandomAccessFile raf = new RandomAccessFile(f, "rw");
+        raf.seek(8);
+        // Mock that there is a ledgers map offset but the ledgers count is 0
+        raf.writeLong(40);
+        raf.writeInt(0);
+        raf.close();
+
+        // now see which ledgers are in the log
+        entryLogger = new DefaultEntryLogger(conf, dirsMgr);
+
+        try {
+            entryLogger.extractEntryLogMetadataFromIndex(0L);
+            fail("Should not be possible to recover from ledgers map index");
+        } catch (IOException e) {
+            assertEquals("No ledgers map found in entryLogId 0, do scan to double confirm", e.getMessage());
+        }
+
+        // Public method should succeed by falling back to scanning the file
+        EntryLogMetadata meta = entryLogger.getEntryLogMetadata(0L);
+        LOG.info("Extracted Meta From Entry Log {}", meta);
+        assertEquals(60, meta.getLedgersMap().get(1L));
+        assertEquals(30, meta.getLedgersMap().get(2L));
+        assertEquals(30, meta.getLedgersMap().get(3L));
+        assertFalse(meta.getLedgersMap().containsKey(4L));
+        assertEquals(120, meta.getTotalSize());
+        assertEquals(120, meta.getRemainingSize());
+    }
+
     /**
      * Explicitly try to recover using the ledgers map index at the end of the entry log.
      */