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

Merge branch 'cassandra-2.0' into cassandra-2.1 

Conflicts:
	src/java/org/apache/cassandra/db/BatchlogManager.java
This commit is contained in:
Aleksey Yeschenko 2014-03-31 13:04:10 +03:00
parent 6844462898 d049017ac8
commit f6daf4e307
2 changed files with 13 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/java/org/apache/cassandra/config/CFMetaData.java
View File

@ -249,7 +249,8 @@ public final class CFMetaData
public static final CFMetaData BatchlogCf = compile("CREATE TABLE " + SystemKeyspace.BATCHLOG_CF + " ("
+ "id uuid PRIMARY KEY,"
+ "written_at timestamp,"
+ "data blob"
+ "data blob,"
+ "version int,"
+ ") WITH COMMENT='uncommited batches' AND gc_grace_seconds=0 "
+ "AND COMPACTION={'class' : 'SizeTieredCompactionStrategy', 'min_threshold' : 2}");

22
src/java/org/apache/cassandra/db/BatchlogManager.java
View File

@ -65,7 +65,6 @@ import org.apache.cassandra.utils.WrappedRunnable;
public class BatchlogManager implements BatchlogManagerMBean
{
private static final String MBEAN_NAME = "org.apache.cassandra.db:type=BatchlogManager";
private static final int VERSION = MessagingService.VERSION_12;
private static final long REPLAY_INTERVAL = 60 * 1000; // milliseconds
private static final int PAGE_SIZE = 128; // same as HHOM, for now, w/out using any heuristics. TODO: set based on avg batch size.
@ -149,7 +148,7 @@ public class BatchlogManager implements BatchlogManagerMBean
{
buf.writeInt(mutations.size());
for (Mutation mutation : mutations)
Mutation.serializer.serialize(mutation, buf, VERSION);
Mutation.serializer.serialize(mutation, buf, MessagingService.VERSION_12);
}
catch (IOException e)
{
@ -174,7 +173,7 @@ public class BatchlogManager implements BatchlogManagerMBean
try
{
UntypedResultSet page = process("SELECT id, data, written_at FROM %s.%s LIMIT %d",
UntypedResultSet page = process("SELECT id, data, written_at, version FROM %s.%s LIMIT %d",
Keyspace.SYSTEM_KS,
SystemKeyspace.BATCHLOG_CF,
PAGE_SIZE);
@ -186,7 +185,7 @@ public class BatchlogManager implements BatchlogManagerMBean
if (page.size() < PAGE_SIZE)
break; // we've exhausted the batchlog, next query would be empty.
page = process("SELECT id, data, written_at FROM %s.%s WHERE token(id) > token(%s) LIMIT %d",
page = process("SELECT id, data, written_at, version FROM %s.%s WHERE token(id) > token(%s) LIMIT %d",
Keyspace.SYSTEM_KS,
SystemKeyspace.BATCHLOG_CF,
id,
@ -211,22 +210,23 @@ public class BatchlogManager implements BatchlogManagerMBean
{
id = row.getUUID("id");
long writtenAt = row.getLong("written_at");
int version = row.has("version") ? row.getInt("version") : MessagingService.VERSION_12;
// enough time for the actual write + batchlog entry mutation delivery (two separate requests).
long timeout = DatabaseDescriptor.getWriteRpcTimeout() * 2; // enough time for the actual write + BM removal mutation
if (System.currentTimeMillis() < writtenAt + timeout)
continue; // not ready to replay yet, might still get a deletion.
replayBatch(id, row.getBytes("data"), writtenAt, rateLimiter);
replayBatch(id, row.getBytes("data"), writtenAt, version, rateLimiter);
}
return id;
}
private void replayBatch(UUID id, ByteBuffer data, long writtenAt, RateLimiter rateLimiter)
private void replayBatch(UUID id, ByteBuffer data, long writtenAt, int version, RateLimiter rateLimiter)
{
logger.debug("Replaying batch {}", id);
try
{
replaySerializedMutations(data, writtenAt, rateLimiter);
replaySerializedMutations(data, writtenAt, version, rateLimiter);
}
catch (IOException e)
{
@ -245,19 +245,19 @@ public class BatchlogManager implements BatchlogManagerMBean
mutation.apply();
}
private void replaySerializedMutations(ByteBuffer data, long writtenAt, RateLimiter rateLimiter) throws IOException
private void replaySerializedMutations(ByteBuffer data, long writtenAt, int version, RateLimiter rateLimiter) throws IOException
{
DataInputStream in = new DataInputStream(ByteBufferUtil.inputStream(data));
int size = in.readInt();
for (int i = 0; i < size; i++)
replaySerializedMutation(Mutation.serializer.deserialize(in, VERSION), writtenAt, rateLimiter);
replaySerializedMutation(Mutation.serializer.deserialize(in, version), writtenAt, version, rateLimiter);
}
/*
* We try to deliver the mutations to the replicas ourselves if they are alive and only resort to writing hints
* when a replica is down or a write request times out.
*/
private void replaySerializedMutation(Mutation mutation, long writtenAt, RateLimiter rateLimiter)
private void replaySerializedMutation(Mutation mutation, long writtenAt, int version, RateLimiter rateLimiter)
{
int ttl = calculateHintTTL(mutation, writtenAt);
if (ttl <= 0)
@ -266,7 +266,7 @@ public class BatchlogManager implements BatchlogManagerMBean
Set<InetAddress> liveEndpoints = new HashSet<>();
String ks = mutation.getKeyspaceName();
Token<?> tk = StorageService.getPartitioner().getToken(mutation.key());
int mutationSize = (int) Mutation.serializer.serializedSize(mutation, VERSION);
int mutationSize = (int) Mutation.serializer.serializedSize(mutation, version);
for (InetAddress endpoint : Iterables.concat(StorageService.instance.getNaturalEndpoints(ks, tk),
StorageService.instance.getTokenMetadata().pendingEndpointsFor(tk, ks)))