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

2.0 compatibility modifications for CASSANDRA-6931 

patch by Aleksey Yeschenko; reviewed by Sylvain Lebresne for
CASSANDRA-6931
This commit is contained in:
Aleksey Yeschenko 2014-03-31 12:53:24 +03:00
parent 6874aaa0db
commit d049017ac8
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

@ -233,7 +233,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

@ -66,7 +66,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.
@ -151,7 +150,7 @@ public class BatchlogManager implements BatchlogManagerMBean
{
out.writeInt(mutations.size());
for (RowMutation rm : mutations)
RowMutation.serializer.serialize(rm, out, VERSION);
RowMutation.serializer.serialize(rm, out, MessagingService.VERSION_12);
}
catch (IOException e)
{
@ -176,7 +175,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);
@ -188,7 +187,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,
@ -213,22 +212,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)
{
@ -247,19 +247,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(RowMutation.serializer.deserialize(in, VERSION), writtenAt, rateLimiter);
replaySerializedMutation(RowMutation.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(RowMutation mutation, long writtenAt, RateLimiter rateLimiter)
private void replaySerializedMutation(RowMutation mutation, long writtenAt, int version, RateLimiter rateLimiter)
{
int ttl = calculateHintTTL(mutation, writtenAt);
if (ttl <= 0)
@ -268,7 +268,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) RowMutation.serializer.serializedSize(mutation, VERSION);
int mutationSize = (int) RowMutation.serializer.serializedSize(mutation, version);
for (InetAddress endpoint : Iterables.concat(StorageService.instance.getNaturalEndpoints(ks, tk),
StorageService.instance.getTokenMetadata().pendingEndpointsFor(tk, ks)))