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/7d8a56df
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/7d8a56df
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/7d8a56df

Branch: refs/heads/trunk
Commit: 7d8a56df3df26be7537f2a5158469629c34b911c
Parents: 71c8912 92b3622
Author: Jonathan Ellis <jbel...@apache.org>
Authored: Thu Oct 3 12:12:11 2013 -0500
Committer: Jonathan Ellis <jbel...@apache.org>
Committed: Thu Oct 3 12:12:11 2013 -0500

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


http://git-wip-us.apache.org/repos/asf/cassandra/blob/7d8a56df/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/service/StorageProxy.java
index 6c8e636,9b559e5..d16bef9
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@@ -963,29 -637,35 +963,29 @@@ 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);
 +                InetAddress destination = iter.next();
 +                CompactEndpointSerializationHelper.serialize(destination, 
out);
-                 int id = MessagingService.instance().addCallback(handler, 
message, destination, message.getTimeout());
++                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);
 +            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());
 -            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)

Reply via email to