From b4210acd3da4d51620a22b20a50ac3bd189cfe42 Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Sun, 27 Jul 2025 23:04:16 +0100 Subject: [PATCH] Improve CommitLogSegmentReader to skip SyncBlocks correctly in case of CRC errors patch by Dmitry Konstantinov; reviewed by Stefan Miklosovic for CASSANDRA-20664 --- CHANGES.txt | 1 + .../db/commitlog/CommitLogSegmentReader.java | 87 ++++++++------ .../db/commitlog/CommitLogReaderTest.java | 107 +++++++++++++++++- 3 files changed, 153 insertions(+), 42 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 60bfe4275c..69ddf14d52 100644 --- a/CHANGES.txt +++ b/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) diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentReader.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentReader.java index 33e70c10ae..ff310fdbef 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentReader.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentReader.java @@ -101,56 +101,69 @@ public class CommitLogSegmentReader implements Iterable 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(); - } - 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); + end = -1; // skip the remaining part of the corrupted log segment } 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); + 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) + { + handleUnrecoverableError(e, !e.invalidCrc && tolerateTruncation); + // if no exception is thrown, the while loop will continue + } + catch (IOException e) + { + 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) diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitLogReaderTest.java b/test/unit/org/apache/cassandra/db/commitlog/CommitLogReaderTest.java index 23440ebfa8..98415a85b4 100644 --- a/test/unit/org/apache/cassandra/db/commitlog/CommitLogReaderTest.java +++ b/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.partitions.PartitionUpdate; 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 class CommitLogReaderTest extends CQLTester @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 class CommitLogReaderTest extends CQLTester 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 @@ public class CommitLogReaderTest extends CQLTester 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 class CommitLogReaderTest extends CQLTester 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 @@ public class CommitLogReaderTest extends CQLTester .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(); + } }