Fix compacted entry log recovery filename

Author: void-ptr974

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

@@ -1470,7 +1470,8 @@ public Collection<CompactionEntryLog> incompleteCompactionLogs() {
                         }
                         continue;
                     }
-                    File finalLogFile = new File(compactedFile.getParentFile(), compactionLogId + ".log");
+                    File finalLogFile = new File(compactedFile.getParentFile(),
+                            logId2HexString(compactionLogId) + ".log");
 
                     compactionLogs.add(
                             new EntryLoggerCompactionEntryLog(compactionLogId, compactedLogId,

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

@@ -48,6 +48,7 @@
 import java.nio.channels.FileChannel;
 import java.nio.charset.Charset;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
@@ -65,6 +66,8 @@
 import java.util.concurrent.locks.Lock;
 import org.apache.bookkeeper.bookie.DefaultEntryLogger.BufferedLogChannel;
 import org.apache.bookkeeper.bookie.LedgerDirsManager.NoWritableLedgerDirException;
+import org.apache.bookkeeper.bookie.storage.CompactionEntryLog;
+import org.apache.bookkeeper.bookie.storage.EntryLogScanner;
 import org.apache.bookkeeper.common.testing.annotations.FlakyTest;
 import org.apache.bookkeeper.conf.ServerConfiguration;
 import org.apache.bookkeeper.conf.TestBKConfiguration;
@@ -983,6 +986,165 @@ public void testReadEntryWithoutLedgerID() throws Exception {
         }
     }
 
+    @Test
+    public void testRecoverCompactedLogUsesHexFinalLogFileNameForLogIdTen() throws Exception {
+        // The next compaction log id is previousLogId + 1. Log id 10 is named "a.log",
+        // so recovery must not rebuild the final log file as the decimal name "10.log".
+        verifyRecoveredCompactionLogUsesExpectedFinalLogFileName(9, "a.log", "10.log");
+    }
+
+    @Test
+    public void testRecoverCompactedLogUsesHexFinalLogFileNameForLogIdSixteen() throws Exception {
+        // Log id 16 is named "10.log". This covers a numeric-looking hex name that
+        // was previously rebuilt as the decimal name "16.log" during recovery.
+        verifyRecoveredCompactionLogUsesExpectedFinalLogFileName(15, "10.log", "16.log");
+    }
+
+    @Test
+    public void testRecoverCompactedLogUsesHexFinalLogFileNameForLargerLogId() throws Exception {
+        // Use a larger id to make the decimal/hex mismatch obvious: 500 decimal is
+        // "1f4" in the entry log file name.
+        verifyRecoveredCompactionLogUsesExpectedFinalLogFileName(499, "1f4.log", "500.log");
+    }
+
+    @Test
+    public void testRecoverCompactedLogKeepsSingleDigitFinalLogFileName() throws Exception {
+        // Single-digit ids have the same decimal and hex file name. Keep this case to
+        // show the recovery path remains compatible when the old and new names match.
+        verifyRecoveredCompactionLogUsesExpectedFinalLogFileName(8, "9.log", null);
+    }
+
+    @Test
+    public void testNewCompactionLogKeepsHexFinalLogFileName() throws Exception {
+        long sourceLogId = 1L;
+        long previousLogId = 499L;
+        long expectedCompactionLogId = previousLogId + 1;
+        String expectedFinalLogFileName = "1f4.log";
+        long ledgerId = 1234L;
+        long entryId = 5L;
+        ByteBuf entry = generateEntry(ledgerId, entryId);
+
+        resetEntryLoggerWithLastLogId(previousLogId);
+
+        // This is the normal, non-recovery compaction path. finalLogFile is derived
+        // from the current compaction log file name, so it should already be hex.
+        CompactionEntryLog compactionLog = entryLogger.newCompactionLog(sourceLogId);
+        assertEquals(expectedCompactionLogId, compactionLog.getDstLogId());
+
+        try {
+            compactionLog.addEntry(ledgerId, entry);
+            compactionLog.flush();
+            // markCompacted() leaves a recovery marker such as
+            // "1f4.log.1.compacted" and removes the temporary compacting file.
+            compactionLog.markCompacted();
+
+            File compactedFile = new File(curDir, expectedFinalLogFileName + "."
+                    + DefaultEntryLogger.logId2HexString(sourceLogId)
+                    + TransactionalEntryLogCompactor.COMPACTED_SUFFIX);
+            File finalLogFile = new File(curDir, expectedFinalLogFileName);
+            assertTrue(compactedFile.exists(), "Compacted recovery marker should exist");
+            assertFalse(finalLogFile.exists(), "Final log file should not exist before makeAvailable");
+
+            // makeAvailable() is the point where the final entry log file is created
+            // on disk. Normal compaction must create "1f4.log", not "500.log".
+            compactionLog.makeAvailable();
+            assertTrue(finalLogFile.exists(), "Normal compaction path should keep the hex-named final log file");
+            assertFalse(new File(curDir, "500.log").exists(),
+                    "Normal compaction path should not create a decimal-named log file");
+
+            assertCompactionLogContainsEntry(compactionLog, ledgerId, entryId);
+        } finally {
+            ReferenceCountUtil.release(entry);
+        }
+    }
+
+    private void verifyRecoveredCompactionLogUsesExpectedFinalLogFileName(long previousLogId,
+                                                                          String expectedFinalLogFileName,
+                                                                          String wrongDecimalLogFileName)
+            throws Exception {
+        long sourceLogId = 1L;
+        long expectedCompactionLogId = previousLogId + 1;
+        long ledgerId = 1234L;
+        long entryId = 5L;
+        ByteBuf entry = generateEntry(ledgerId, entryId);
+
+        // Force the next compaction log id to a known value. This lets the test
+        // control both the compacted marker name and the expected final log name.
+        resetEntryLoggerWithLastLogId(previousLogId);
+
+        CompactionEntryLog compactionLog = entryLogger.newCompactionLog(sourceLogId);
+        assertEquals(expectedCompactionLogId, compactionLog.getDstLogId());
+
+        try {
+            compactionLog.addEntry(ledgerId, entry);
+            compactionLog.flush();
+            // Simulate a restart after the compaction log was marked compacted but
+            // before makeAvailable()/index recovery finished. The remaining file is
+            // named "<dst-hex>.log.<src-hex>.compacted".
+            compactionLog.markCompacted();
+
+            File compactedFile = new File(curDir, expectedFinalLogFileName + "."
+                    + DefaultEntryLogger.logId2HexString(sourceLogId)
+                    + TransactionalEntryLogCompactor.COMPACTED_SUFFIX);
+            assertTrue(compactedFile.exists(), "Compacted recovery marker should exist");
+            assertFalse(new File(curDir, expectedFinalLogFileName).exists(),
+                    "Final log file should not exist before recovery");
+
+            // incompleteCompactionLogs() rebuilds EntryLoggerCompactionEntryLog from
+            // the marker. The bug was here: the rebuilt finalLogFile used the decimal
+            // compactionLogId instead of the original hex entry log file name.
+            Collection<CompactionEntryLog> incompleteLogs = entryLogger.incompleteCompactionLogs();
+            assertEquals(1, incompleteLogs.size());
+
+            CompactionEntryLog recoveredLog = incompleteLogs.iterator().next();
+            assertEquals(expectedCompactionLogId, recoveredLog.getDstLogId());
+            assertEquals(sourceLogId, recoveredLog.getSrcLogId());
+
+            // Recovery creates the final entry log when makeAvailable() is called.
+            // It must create the hex name and must not leave a decimal-named log.
+            recoveredLog.makeAvailable();
+            assertTrue(new File(curDir, expectedFinalLogFileName).exists(),
+                    "Recovery should make the hex-named log file available");
+            if (wrongDecimalLogFileName != null) {
+                assertFalse(new File(curDir, wrongDecimalLogFileName).exists(),
+                        "Recovery should not create a decimal-named log file");
+            }
+
+            assertCompactionLogContainsEntry(recoveredLog, ledgerId, entryId);
+        } finally {
+            ReferenceCountUtil.release(entry);
+        }
+    }
+
+    private void assertCompactionLogContainsEntry(CompactionEntryLog compactionLog,
+                                                  long ledgerId,
+                                                  long entryId) throws Exception {
+        AtomicBoolean foundEntry = new AtomicBoolean(false);
+        // Scanning uses the compaction log id to resolve the final log file. This
+        // catches cases where makeAvailable() created a file under the wrong name.
+        compactionLog.scan(new EntryLogScanner() {
+            @Override
+            public boolean accept(long ledgerId) {
+                return true;
+            }
+
+            @Override
+            public void process(long ledgerIdFromScanner, long offset, ByteBuf recoveredEntry) {
+                assertEquals(ledgerId, ledgerIdFromScanner);
+                assertEquals(ledgerId, recoveredEntry.getLong(recoveredEntry.readerIndex()));
+                assertEquals(entryId, recoveredEntry.getLong(recoveredEntry.readerIndex() + Long.BYTES));
+                foundEntry.set(true);
+            }
+        });
+        assertTrue(foundEntry.get(), "Compacted log should be readable through the compaction log id");
+    }
+
+    private void resetEntryLoggerWithLastLogId(long lastLogId) throws Exception {
+        entryLogger.getEntryLoggerAllocator().setLastLogId(curDir, lastLogId);
+        entryLogger.close();
+        entryLogger = new DefaultEntryLogger(conf, dirsMgr);
+    }
+
 
     /*
      * tests basic logic of EntryLogManager interface for