Improve CommitLogSegmentReader to skip SyncBlocks correctly in case of CRC errors

patch by Dmitry Konstantinov; reviewed by Stefan Miklosovic for CASSANDRA-20664

Author: Dmitry Konstantinov

CHANGES.txt

@@ -1,4 +1,5 @@
 4.1.10
+ * Improve CommitLogSegmentReader to skip SyncBlocks correctly in case of CRC errors (CASSANDRA-20664)
  * Do not crash on first boot with data_disk_usage_max_disk_size set when data directory is not created yet (CASSANDRA-20787)
  * Rework / simplification of nodetool get/setguardrailsconfig commands (CASSANDRA-20778)
  * IntrusiveStack.accumulate is not accumulating correctly (CASSANDRA-20670)

src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentReader.java

@@ -101,56 +101,69 @@ protected SyncSegment computeNext()
         {
             while (true)
             {
+                final int currentStart = end;
                 try
                 {
-                    final int currentStart = end;
                     end = readSyncMarker(descriptor, currentStart, reader);
-                    if (end == -1)
-                    {
-                        return endOfData();
-                    }
-                    if (end > reader.length())
-                    {
-                        // the CRC was good (meaning it was good when it was written and still looks legit), but the file is truncated now.
-                        // try to grab and use as much of the file as possible, which might be nothing if the end of the file truly is corrupt
-                        end = (int) reader.length();
-                    }
+                }
+                catch (CommitLogSegmentReader.SegmentReadException e)
+                {
+                    handleUnrecoverableError(e, !e.invalidCrc && tolerateTruncation);
+                    end = -1; // skip the remaining part of the corrupted log segment
+                }
+                catch (IOException e)
+                {
+                    boolean tolerateErrorsInSection = tolerateTruncation & segmenter.tolerateSegmentErrors(end, reader.length());
+                    handleUnrecoverableError(e, tolerateErrorsInSection);
+                    end = -1; // skip the remaining part of the corrupted log segment
+                }
+
+                if (end == -1)
+                {
+                    return endOfData();
+                }
+                if (end > reader.length())
+                {
+                    // the CRC was good (meaning it was good when it was written and still looks legit), but the file is truncated now.
+                    // try to grab and use as much of the file as possible, which might be nothing if the end of the file truly is corrupt
+                    end = (int) reader.length();
+                }
+
+                try
+                {
                     return segmenter.nextSegment(currentStart + SYNC_MARKER_SIZE, end);
                 }
-                catch(CommitLogSegmentReader.SegmentReadException e)
+                catch (CommitLogSegmentReader.SegmentReadException e)
                 {
-                    try
-                    {
-                        handler.handleUnrecoverableError(new CommitLogReadException(
-                                                    e.getMessage(),
-                                                    CommitLogReadErrorReason.UNRECOVERABLE_DESCRIPTOR_ERROR,
-                                                    !e.invalidCrc && tolerateTruncation));
-                    }
-                    catch (IOException ioe)
-                    {
-                        throw new RuntimeException(ioe);
-                    }
+                    handleUnrecoverableError(e, !e.invalidCrc && tolerateTruncation);
+                    // if no exception is thrown, the while loop will continue
                 }
                 catch (IOException e)
                 {
-                    try
-                    {
-                        boolean tolerateErrorsInSection = tolerateTruncation & segmenter.tolerateSegmentErrors(end, reader.length());
-                        // if no exception is thrown, the while loop will continue
-                        handler.handleUnrecoverableError(new CommitLogReadException(
-                                                    e.getMessage(),
-                                                    CommitLogReadErrorReason.UNRECOVERABLE_DESCRIPTOR_ERROR,
-                                                    tolerateErrorsInSection));
-                    }
-                    catch (IOException ioe)
-                    {
-                        throw new RuntimeException(ioe);
-                    }
+                    boolean tolerateErrorsInSection = tolerateTruncation & segmenter.tolerateSegmentErrors(end, reader.length());
+                    handleUnrecoverableError(e, tolerateErrorsInSection);
+                    // if no exception is thrown, the while loop will continue
                 }
             }
         }
     }
 
+    private void handleUnrecoverableError(Exception e, boolean permissible)
+    {
+        try
+        {
+            handler.handleUnrecoverableError(new CommitLogReadException(
+                e.getMessage(),
+                CommitLogReadErrorReason.UNRECOVERABLE_DESCRIPTOR_ERROR,
+                permissible)
+            );
+        }
+        catch (IOException ioe)
+        {
+            throw new RuntimeException(ioe);
+        }
+    }
+
     private int readSyncMarker(CommitLogDescriptor descriptor, int offset, RandomAccessReader reader) throws IOException
     {
         if (offset > reader.length() - SYNC_MARKER_SIZE)

test/unit/org/apache/cassandra/db/commitlog/CommitLogReaderTest.java

@@ -17,11 +17,15 @@
  */
 package org.apache.cassandra.db.commitlog;
 
+import java.io.FileOutputStream;
 import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
 
+import org.apache.cassandra.distributed.shared.WithProperties;
 import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.security.EncryptionContextGenerator;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -40,15 +44,20 @@
 import org.apache.cassandra.db.rows.Row;
 import org.apache.cassandra.utils.JVMStabilityInspector;
 import org.apache.cassandra.utils.KillerForTests;
+import org.assertj.core.api.Assertions;
+
+import static org.apache.cassandra.db.commitlog.CommitLogReplayer.IGNORE_REPLAY_ERRORS_PROPERTY;
 
 public class CommitLogReaderTest extends CQLTester
 {
+    private static final long CORRUPTED_COMMIT_LOG_FILE_ID = 111L;
+    private static final String CORRUPTED_COMMIT_LOG_FILE_NAME = "CommitLog-7-1234567.log";
+
     @BeforeClass
     public static void setUpClass()
     {
         prePrepareServer();
 
-        DatabaseDescriptor.setCommitFailurePolicy(Config.CommitFailurePolicy.ignore);
         JVMStabilityInspector.replaceKiller(new KillerForTests(false));
 
         DatabaseDescriptor.setCommitLogSync(Config.CommitLogSync.batch);
@@ -60,7 +69,12 @@ public static void setUpClass()
     @Before
     public void before() throws IOException
     {
+        clearCorruptedCommitLogFile();
         CommitLog.instance.resetUnsafe(true);
+
+        // always reset to what Cassandra's default is and let each test method
+        // handle its expected failure policy itself for better test encapsulation.
+        DatabaseDescriptor.setCommitFailurePolicy(Config.CommitFailurePolicy.stop);
     }
 
     @Test
@@ -165,6 +179,58 @@ public void testReadCountFromMidpoint() throws Throwable
         confirmReadOrder(testHandler, samples / 2);
     }
 
+    @Test
+    public void testSyncMarkerChecksumReadFailed_ignoreReplayErrorsDisabled() throws Throwable
+    {
+        File corruptedSegmentFile = createAndWriteCorruptedCommitLogFile();
+        CommitLogReader reader = new CommitLogReader();
+        // use real CLR handler to test actual behavior
+        CommitLogReadHandler clrHandler =
+                new CommitLogReplayer(new CommitLog(null), null, null, null);
+
+        // ignore.replay.errors disabled, so we expect the exception here
+        Assertions.assertThatThrownBy(() ->
+                                      reader.readCommitLogSegment(clrHandler,
+                                                                  corruptedSegmentFile,
+                                                                  CommitLogPosition.NONE,
+                                                                  CommitLogReader.ALL_MUTATIONS,
+                                                                  false)
+                  ).isInstanceOf(CommitLogReplayer.CommitLogReplayException.class);
+    }
+
+    @Test
+    public void testSyncMarkerChecksumReadFailed_ignoreReplayErrorsEnabled() throws Throwable
+    {
+        try (WithProperties properties = new WithProperties(IGNORE_REPLAY_ERRORS_PROPERTY, "true"))
+        {
+            File corruptedSegmentFile = createAndWriteCorruptedCommitLogFile();
+
+            CommitLogReader reader = new CommitLogReader();
+            // use real CLR handler to test actual behavior
+            CommitLogReadHandler clrHandler =
+            new CommitLogReplayer(new CommitLog(null), null, null, null);
+
+            // ignore.replay.errors enabled, so we don't expect any errors
+            reader.readCommitLogSegment(clrHandler, corruptedSegmentFile, CommitLogPosition.NONE, CommitLogReader.ALL_MUTATIONS, false);
+        }
+    }
+
+    @Test
+    public void testSyncMarkerChecksumReadFailed_ignoreReplayErrorsDisabled_commitFailurePolicyIgnore() throws Throwable
+    {
+        DatabaseDescriptor.setCommitFailurePolicy(Config.CommitFailurePolicy.ignore);
+
+        File corruptedSegmentFile = createAndWriteCorruptedCommitLogFile();
+
+        CommitLogReader reader = new CommitLogReader();
+        // use real CLR handler to test actual behavior
+        CommitLogReadHandler clrHandler =
+                new CommitLogReplayer(new CommitLog(null), null, null, null);
+
+        // commit.failure.policy=ignore, so we don't expect any errors
+        reader.readCommitLogSegment(clrHandler, corruptedSegmentFile, CommitLogPosition.NONE, CommitLogReader.ALL_MUTATIONS, false);
+    }
+
     /**
      * Since we have both table and non mixed into the CL, we ignore updates that aren't for the table the test handler
      * is configured to check.
@@ -207,7 +273,7 @@ static ArrayList<File> getCommitLogs()
                 continue;
             results.add(f);
         }
-        Assert.assertTrue("Didn't find any commit log files.", 0 != results.size());
+        Assert.assertFalse("Didn't find any commit log files.", results.isEmpty());
         return results;
     }
 
@@ -229,20 +295,20 @@ public TestCLRHandler(TableMetadata metadata)
             this.metadata = metadata;
         }
 
-        public boolean shouldSkipSegmentOnError(CommitLogReadException exception) throws IOException
+        public boolean shouldSkipSegmentOnError(CommitLogReadException exception)
         {
             sawStopOnErrorCheck = true;
             return false;
         }
 
-        public void handleUnrecoverableError(CommitLogReadException exception) throws IOException
+        public void handleUnrecoverableError(CommitLogReadException exception)
         {
             sawStopOnErrorCheck = true;
         }
 
         public void handleMutation(Mutation m, int size, int entryLocation, CommitLogDescriptor desc)
         {
-            if ((metadata == null) || (metadata != null && m.getPartitionUpdate(metadata) != null)) {
+            if (metadata == null || m.getPartitionUpdate(metadata) != null) {
                 seenMutations.add(m);
             }
         }
@@ -274,4 +340,35 @@ CommitLogPosition populateData(int entryCount) throws Throwable
                 .forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
         return result;
     }
+
+    private static File createAndWriteCorruptedCommitLogFile() throws IOException
+    {
+        final ByteBuffer corruptedSegmentByteBuffer =
+                ByteBuffer.allocate(DatabaseDescriptor.getCommitLogSegmentSize());
+
+        final CommitLogDescriptor commitLogDescriptor =
+                new CommitLogDescriptor(CORRUPTED_COMMIT_LOG_FILE_ID, null, EncryptionContextGenerator.createDisabledContext());
+
+        CommitLogDescriptor.writeHeader(corruptedSegmentByteBuffer, commitLogDescriptor);
+
+        // write corrupted sync marker:
+        // put wrong offset
+        corruptedSegmentByteBuffer.putInt(42);
+        // put wrong CRC
+        corruptedSegmentByteBuffer.putInt(42);
+
+        final File corruptedLogFile = new File(DatabaseDescriptor.getCommitLogLocation(), CORRUPTED_COMMIT_LOG_FILE_NAME);
+        try (FileOutputStream fos = new FileOutputStream(corruptedLogFile.toJavaIOFile()))
+        {
+            fos.write(corruptedSegmentByteBuffer.array());
+            fos.flush();
+        }
+
+        return corruptedLogFile;
+    }
+
+    private static void clearCorruptedCommitLogFile()
+    {
+        new File(DatabaseDescriptor.getCommitLogLocation(), CORRUPTED_COMMIT_LOG_FILE_NAME).deleteIfExists();
+    }
 }