Issue #2220: support skip invalid record in recovery

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

@@ -618,7 +618,7 @@ private void replay(Journal journal, JournalScanner scanner) throws IOException
                 logPosition = markedLog.getLogFileOffset();
             }
             LOG.info("Replaying journal {} from position {}", id, logPosition);
-            long scanOffset = journal.scanJournal(id, logPosition, scanner);
+            long scanOffset = journal.scanJournal(id, logPosition, scanner, this.conf.isSkipReplayJournalInvalidRecord());
             // Update LastLogMark after completely replaying journal
             // scanOffset will point to EOF position
             // After LedgerStorage flush, SyncThread should persist this to disk

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

@@ -848,13 +848,14 @@ public void checkpointComplete(Checkpoint checkpoint, boolean compact) throws IO
     /**
      * Scan the journal.
      *
-     * @param journalId Journal Log Id
-     * @param journalPos Offset to start scanning
-     * @param scanner Scanner to handle entries
+     * @param journalId         Journal Log Id
+     * @param journalPos        Offset to start scanning
+     * @param scanner           Scanner to handle entries
+     * @param skipInvalidRecord when invalid record,should we skip it or not
      * @return scanOffset - represents the byte till which journal was read
      * @throws IOException
      */
-    public long scanJournal(long journalId, long journalPos, JournalScanner scanner)
+    public long scanJournal(long journalId, long journalPos, JournalScanner scanner, boolean skipInvalidRecord)
         throws IOException {
         JournalChannel recLog;
         if (journalPos <= 0) {
@@ -897,7 +898,12 @@ public long scanJournal(long journalId, long journalPos, JournalScanner scanner)
                             continue;
                         }
                         isPaddingRecord = true;
-                    } else {
+                    } else if (skipInvalidRecord){
+                        LOG.warn("Invalid record found with negative length: {},because of " +
+                                "skipInvalidRecord is true,we skip the next data", len);
+                        break;
+                    }
+                    else {
                         LOG.error("Invalid record found with negative length: {}", len);
                         throw new IOException("Invalid record found with negative length " + len);
                     }

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

@@ -223,7 +223,7 @@ public void process(int journalVersion, long offset, ByteBuffer entry) {
                     LOG.info("Found ledger {} in journal", ledgerId);
                 }
             }
-        });
+        }, false);
     }
 
     private void delete(Path path) {

bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java

@@ -334,6 +334,8 @@ public class ServerConfiguration extends AbstractConfiguration<ServerConfigurati
     // Used for location index, lots of writes and much bigger dataset
     protected static final String LEDGER_METADATA_ROCKSDB_CONF = "ledgerMetadataRocksdbConf";
 
+    protected static final String SKIP_REPLAY_JOURNAL_INVALID_RECORD = "skipReplayJournalInvalidRecord";
+
     /**
      * Construct a default configuration object.
      */
@@ -3917,6 +3919,25 @@ public boolean isDataIntegrityStampMissingCookiesEnabled() {
         return this.getBoolean(DATA_INTEGRITY_COOKIE_STAMPING_ENABLED, false);
     }
 
+
+    /**
+     * When this config is set to true,if we replay journal failed, we will skip
+     * @param skipReplayJournalInvalidRecord
+     * @return
+     */
+    public ServerConfiguration setSkipReplayJournalInvalidRecord(boolean skipReplayJournalInvalidRecord) {
+        this.setProperty(SKIP_REPLAY_JOURNAL_INVALID_RECORD,
+                Boolean.toString(skipReplayJournalInvalidRecord));
+        return this;
+    }
+
+    /**
+     * @see #isSkipReplayJournalInvalidRecord
+     */
+    public boolean isSkipReplayJournalInvalidRecord() {
+        return this.getBoolean(SKIP_REPLAY_JOURNAL_INVALID_RECORD, false);
+    }
+
     /**
      * Get default rocksdb conf.
      *

bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/bookie/ReadJournalCommand.java

@@ -211,6 +211,6 @@ public void process(int journalVersion, long offset, ByteBuffer entry) throws IO
     }
 
     private void scanJournal(Journal journal, long journalId, Journal.JournalScanner scanner) throws IOException {
-        journal.scanJournal(journalId, 0L, scanner);
+        journal.scanJournal(journalId, 0L, scanner, false);
     }
 }

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

@@ -299,6 +299,51 @@ private JournalChannel writeV4Journal(File journalDir, int numEntries, byte[] ma
         return jc;
     }
 
+    private JournalChannel writeV4JournalWithInvalidRecord(File journalDir, int numEntries, byte[] masterKey) throws Exception {
+        long logId = System.currentTimeMillis();
+        JournalChannel jc = new JournalChannel(journalDir, logId);
+
+        moveToPosition(jc, JournalChannel.VERSION_HEADER_SIZE);
+
+        BufferedChannel bc = jc.getBufferedChannel();
+
+        byte[] data = new byte[1024];
+        Arrays.fill(data, (byte) 'X');
+        long lastConfirmed = LedgerHandle.INVALID_ENTRY_ID;
+        for (int i = 0; i <= numEntries; i++) {
+            ByteBuf packet;
+            if (i == 0) {
+                packet = generateMetaEntry(1, masterKey);
+            } else {
+                packet = ClientUtil.generatePacket(1, i, lastConfirmed, i * data.length, data);
+            }
+            lastConfirmed = i;
+            ByteBuffer lenBuff = ByteBuffer.allocate(4);
+            if (i == numEntries - 1) {
+                //mock when flush data to file ,it writes an invalid entry to journal
+                lenBuff.putInt(-1);
+            } else {
+                lenBuff.putInt(packet.readableBytes());
+            }
+            lenBuff.flip();
+            bc.write(Unpooled.wrappedBuffer(lenBuff));
+            bc.write(packet);
+            packet.release();
+        }
+
+        // write fence key
+        ByteBuf packet = generateFenceEntry(1);
+        ByteBuf lenBuf = Unpooled.buffer();
+        lenBuf.writeInt(packet.readableBytes());
+        //mock
+        bc.write(lenBuf);
+        bc.write(packet);
+        bc.flushAndForceWrite(false);
+        updateJournalVersion(jc, JournalChannel.V4);
+
+        return jc;
+    }
+
     static JournalChannel writeV5Journal(File journalDir, int numEntries,
                                          byte[] masterKey) throws Exception {
         return writeV5Journal(journalDir, numEntries, masterKey, false);
@@ -844,7 +889,7 @@ public void testJournalScanIOException() throws Exception {
             assertEquals(journalIds.size(), 1);
 
             try {
-                journal.scanJournal(journalIds.get(0), Long.MAX_VALUE, journalScanner);
+                journal.scanJournal(journalIds.get(0), Long.MAX_VALUE, journalScanner, false);
                 fail("Should not have been able to scan the journal");
             } catch (Exception e) {
                 // Expected
@@ -854,6 +899,46 @@ public void testJournalScanIOException() throws Exception {
         b.shutdown();
     }
 
+    /**
+     * Test for invalid record data during read of Journal.
+     */
+    @Test
+    public void testJournalScanInvalidRecordWithSkipFlag() throws Exception {
+        File journalDir = createTempDir("bookie", "journal");
+        BookieImpl.checkDirectoryStructure(BookieImpl.getCurrentDirectory(journalDir));
+
+        File ledgerDir = createTempDir("bookie", "ledger");
+        BookieImpl.checkDirectoryStructure(BookieImpl.getCurrentDirectory(ledgerDir));
+
+        writeV4JournalWithInvalidRecord(BookieImpl.getCurrentDirectory(journalDir), 100, "testPasswd".getBytes());
+
+
+        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
+        conf.setJournalDirName(journalDir.getPath())
+                .setLedgerDirNames(new String[] { ledgerDir.getPath() })
+                .setMetadataServiceUri(null)
+                .setSkipReplayJournalInvalidRecord(true);
+
+        Journal.JournalScanner journalScanner = new DummyJournalScan();
+
+        BookieImpl b = new TestBookieImpl(conf);
+
+        for (Journal journal : b.journals) {
+            List<Long> journalIds = journal.listJournalIds(journal.getJournalDirectory(), null);
+
+            assertEquals(journalIds.size(), 1);
+
+            try {
+                journal.scanJournal(journalIds.get(0), 0, journalScanner, conf.isSkipReplayJournalInvalidRecord());
+            } catch (Exception e) {
+                fail("Should have been able to scan the journal,because of skip flag");
+            }
+        }
+
+        b.shutdown();
+    }
+
+
     private class DummyJournalScan implements Journal.JournalScanner {
 
         @Override