natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Improve CommitLogSegmentReader to skip SyncBlocks correctly in case of CRC errors 

patch by Dmitry Konstantinov; reviewed by Stefan Miklosovic for CASSANDRA-20664
This commit is contained in:
Dmitry Konstantinov 2025-07-27 23:04:16 +01:00 committed by Stefan Miklosovic
parent fcaa1b3e39
commit b4210acd3d
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
3 changed files with 153 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

87
src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentReader.java
View File

@ -101,56 +101,69 @@ public class CommitLogSegmentReader implements Iterable<CommitLogSegmentReader.S
{
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();
}
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)

107
test/unit/org/apache/cassandra/db/commitlog/CommitLogReaderTest.java
View File

@ -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();
}
}