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/cassandra-2.1 Commit: 2c7622a65ce747819931bd52bc576a4cd055ba3d Parents: a16adba b9324e1 Author: Aleksey Yeschenko <alek...@apache.org> Authored: Tue Apr 22 22:16:27 2014 +0300 Committer: Aleksey Yeschenko <alek...@apache.org> 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)