Merge branch 'cassandra-2.0' into cassandra-2.1
Conflicts:
src/java/org/apache/cassandra/service/StorageProxy.java
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/2c7622a6
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/2c7622a6
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/2c7622a6
Branch: refs/heads/trunk
Commit: 2c7622a65ce747819931bd52bc576a4cd055ba3d
Parents: a16adba b9324e1
Author: Aleksey Yeschenko <[email protected]>
Authored: Tue Apr 22 22:16:27 2014 +0300
Committer: Aleksey Yeschenko <[email protected]>
Committed: Tue Apr 22 22:16:27 2014 +0300
----------------------------------------------------------------------
src/java/org/apache/cassandra/service/StorageProxy.java | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2c7622a6/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/service/StorageProxy.java
index 33f6ff0,8196352..d8c5813
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@@ -642,15 -620,22 +642,15 @@@ public class StorageProxy implements St
Keyspace.open(Keyspace.SYSTEM_KS),
null,
WriteType.SIMPLE);
- RowMutation rm = new RowMutation(Keyspace.SYSTEM_KS,
UUIDType.instance.decompose(uuid), cf);
- updateBatchlog(rm, endpoints, handler);
- }
-
- private static void updateBatchlog(RowMutation rm,
Collection<InetAddress> endpoints, AbstractWriteResponseHandler handler)
- {
- if (endpoints.contains(FBUtilities.getBroadcastAddress()))
- {
- assert endpoints.size() == 1;
- insertLocal(rm, handler);
- }
- else
+ Mutation mutation = new Mutation(Keyspace.SYSTEM_KS,
UUIDType.instance.decompose(uuid));
+ mutation.delete(SystemKeyspace.BATCHLOG_CF,
FBUtilities.timestampMicros());
+ MessageOut<Mutation> message = mutation.createMessage();
+ for (InetAddress target : endpoints)
{
- MessageOut<RowMutation> message = rm.createMessage();
- for (InetAddress target : endpoints)
+ if (target.equals(FBUtilities.getBroadcastAddress()) &&
OPTIMIZE_LOCAL_REQUESTS)
+ insertLocal(message.payload, handler);
+ else
- MessagingService.instance().sendRR(message, target, handler);
+ MessagingService.instance().sendRR(message, target, handler,
false);
}
}
@@@ -823,9 -812,9 +823,9 @@@
String dc =
DatabaseDescriptor.getEndpointSnitch().getDatacenter(destination);
// direct writes to local DC or old Cassandra versions
// (1.1 knows how to forward old-style String message
IDs; updated to int in 2.0)
- if (localDataCenter.equals(dc) ||
MessagingService.instance().getVersion(destination) <
MessagingService.VERSION_20)
+ if (localDataCenter.equals(dc))
{
- MessagingService.instance().sendRR(message,
destination, responseHandler);
+ MessagingService.instance().sendRR(message,
destination, responseHandler, true);
}
else
{
@@@ -946,9 -935,9 +946,9 @@@
out.writeInt(id);
logger.trace("Adding FWD message to {}@{}", id, destination);
}
- message = message.withParameter(RowMutation.FORWARD_TO,
out.getData());
+ message = message.withParameter(Mutation.FORWARD_TO,
out.getData());
// send the combined message + forward headers
- int id = MessagingService.instance().sendRR(message, target,
handler);
+ int id = MessagingService.instance().sendRR(message, target,
handler, true);
logger.trace("Sending message to {}@{}", id, target);
}
catch (IOException e)
