cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject [5/6] git commit: merge from 1.2
Date Fri, 04 Oct 2013 03:18:19 GMT
merge from 1.2


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/ef69bd53
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/ef69bd53
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/ef69bd53

Branch: refs/heads/cassandra-2.0
Commit: ef69bd532f066a7b78f10db093e266b3eff90ed6
Parents: ff7d341 adb3a0d
Author: Jonathan Ellis <jbellis@apache.org>
Authored: Thu Oct 3 22:17:49 2013 -0500
Committer: Jonathan Ellis <jbellis@apache.org>
Committed: Thu Oct 3 22:17:49 2013 -0500

----------------------------------------------------------------------
 src/java/org/apache/cassandra/service/StorageProxy.java | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/ef69bd53/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/service/StorageProxy.java
index d16bef9,ec33723..8d01281
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@@ -963,29 -637,35 +963,30 @@@ public class StorageProxy implements St
          Iterator<InetAddress> iter = targets.iterator();
          InetAddress target = iter.next();
  
 -        // direct writes to local DC or old Cassandra versions
 -        if (localDC || MessagingService.instance().getVersion(target) < MessagingService.VERSION_11)
 +        // Add the other destinations of the same message as a FORWARD_HEADER entry
 +        DataOutputBuffer out = new DataOutputBuffer();
 +        try
          {
 -            // yes, the loop and non-loop code here are the same; this is clunky but we
want to avoid
 -            // creating a second iterator since we already have a perfectly good one
 -            MessagingService.instance().sendRR(message, target, handler, message.getTimeout(),
handler.consistencyLevel);
 +            out.writeInt(targets.size() - 1);
              while (iter.hasNext())
              {
 -                target = iter.next();
 -                MessagingService.instance().sendRR(message, target, handler, message.getTimeout(),
handler.consistencyLevel);
 +                InetAddress destination = iter.next();
 +                CompactEndpointSerializationHelper.serialize(destination, out);
 +                int id = MessagingService.instance().addCallback(handler, message, destination,
message.getTimeout(), handler.consistencyLevel);
 +                out.writeInt(id);
 +                logger.trace("Adding FWD message to {}@{}", id, destination);
              }
 -            return;
 +            message = message.withParameter(RowMutation.FORWARD_TO, out.getData());
 +            // send the combined message + forward headers
-             int id = MessagingService.instance().sendRR(message, target, handler);
++            int id = MessagingService.instance().addCallback(handler, message, target, message.getTimeout(),
handler.consistencyLevel);
++            MessagingService.instance().sendOneWay(message, id, target);
 +            logger.trace("Sending message to {}@{}", id, target);
          }
 -
 -        // Add all the other destinations of the same message as a FORWARD_HEADER entry
 -        FastByteArrayOutputStream bos = new FastByteArrayOutputStream();
 -        DataOutputStream dos = new DataOutputStream(bos);
 -        dos.writeInt(targets.size() - 1);
 -        while (iter.hasNext())
 +        catch (IOException e)
          {
 -            InetAddress destination = iter.next();
 -            CompactEndpointSerializationHelper.serialize(destination, dos);
 -            String id = MessagingService.instance().addCallback(handler, message, destination,
message.getTimeout(), handler.consistencyLevel);
 -            dos.writeUTF(id);
 +            // DataOutputBuffer is in-memory, doesn't throw IOException
 +            throw new AssertionError(e);
          }
 -        message = message.withParameter(RowMutation.FORWARD_TO, bos.toByteArray());
 -        // send the combined message + forward headers
 -        Tracing.trace("Enqueuing message to {}", target);
 -        MessagingService.instance().sendRR(message, target, handler);
      }
  
      private static void insertLocal(final RowMutation rm, final AbstractWriteResponseHandler
responseHandler)


Mime
View raw message