diff --git a/CHANGES.txt b/CHANGES.txt index e0863eb85a..700ebd319c 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,6 @@ 0.5.0 RC1 * Fix potential NPE in get_range_slice (CASSANDRA-623) + * add CRC32 to commitlog entries (CASSANDRA-605) 0.5.0 beta 2 diff --git a/NEWS.txt b/NEWS.txt index 9ce55745f8..1cf37829c3 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -6,6 +6,7 @@ or simply remove the commitlog directory if you only have test data. (If more writes come in after the flush command, starting 0.5 will error out; if that happens, just go back to 0.4 and flush again.) + The format changed twice: from 0.4 to beta1, and from beta2 to RC1. 1. Bootstrap, move, load balancing, and active repair have been added. See http://wiki.apache.org/cassandra/Operations diff --git a/src/java/org/apache/cassandra/db/CommitLog.java b/src/java/org/apache/cassandra/db/CommitLog.java index 09725b215c..a28e1ed83f 100644 --- a/src/java/org/apache/cassandra/db/CommitLog.java +++ b/src/java/org/apache/cassandra/db/CommitLog.java @@ -31,6 +31,8 @@ import org.apache.log4j.Logger; import java.io.*; import java.util.*; +import java.util.zip.Checksum; +import java.util.zip.CRC32; import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; @@ -305,11 +307,13 @@ public class CommitLog if (logger_.isDebugEnabled()) logger_.debug("Reading mutation at " + reader.getFilePointer()); + long claimedCRC32; byte[] bytes; try { bytes = new byte[(int) reader.readLong()]; // readlong can throw EOFException too reader.readFully(bytes); + claimedCRC32 = reader.readLong(); } catch (EOFException e) { @@ -317,8 +321,16 @@ public class CommitLog break; } bufIn.reset(bytes, bytes.length); + Checksum checksum = new CRC32(); + checksum.update(bytes, 0, bytes.length); + if (claimedCRC32 != checksum.getValue()) + { + // this part of the log must not have been fsynced. probably the rest is bad too, + // but just in case there is no harm in trying them. + continue; + } - /* read the commit log entry */ + /* deserialize the commit log entry */ final RowMutation rm = RowMutation.serializer().deserialize(bufIn); if (logger_.isDebugEnabled()) logger_.debug(String.format("replaying mutation for %s.%s: %s", @@ -620,16 +632,16 @@ public class CommitLog long currentPosition = -1L; try { - /* serialize the row */ currentPosition = logWriter_.getFilePointer(); CommitLogContext cLogCtx = new CommitLogContext(logFile_, currentPosition); - /* Update the header */ maybeUpdateHeader(rowMutation); + Checksum checkum = new CRC32(); if (serializedRow instanceof DataOutputBuffer) { DataOutputBuffer buffer = (DataOutputBuffer) serializedRow; logWriter_.writeLong(buffer.getLength()); logWriter_.write(buffer.getData(), 0, buffer.getLength()); + checkum.update(buffer.getData(), 0, buffer.getLength()); } else { @@ -637,7 +649,9 @@ public class CommitLog byte[] bytes = (byte[]) serializedRow; logWriter_.writeLong(bytes.length); logWriter_.write(bytes); + checkum.update(bytes, 0, bytes.length); } + logWriter_.writeLong(checkum.getValue()); maybeRollLog(); return cLogCtx; }