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

merge from 0.5 branch 

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@889662 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-12-11 15:32:05 +00:00
parent 0967566af3
commit d4d8381daa
3 changed files with 19 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

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

1
NEWS.txt
View File

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

20
src/java/org/apache/cassandra/db/CommitLog.java
View File

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