cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jasobr...@apache.org
Subject [04/11] cassandra git commit: move streaming to use netty
Date Tue, 22 Aug 2017 20:55:11 GMT
move streaming to use netty

patch by jasobrown, reviewed by aweisberg for CASSANDRA-12229


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

Branch: refs/heads/trunk
Commit: fc92db2b9b56c143516026ba29cecdec37e286bb
Parents: 356dc3c
Author: Jason Brown <jasedbrown@gmail.com>
Authored: Mon Apr 11 05:26:18 2016 -0700
Committer: Jason Brown <jasedbrown@gmail.com>
Committed: Tue Aug 22 13:54:44 2017 -0700

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 lib/compress-lzf-0.8.4.jar                      | Bin 25490 -> 0 bytes
 .../org/apache/cassandra/config/Config.java     |   6 -
 .../cassandra/config/DatabaseDescriptor.java    |  15 -
 .../exceptions/ChecksumMismatchException.java   |  34 ++
 .../io/compress/CompressionMetadata.java        |   4 +-
 .../cassandra/io/sstable/SSTableLoader.java     |   2 +-
 .../io/util/DataIntegrityMetadata.java          |  26 +
 .../net/IncomingStreamingConnection.java        | 104 ----
 .../net/async/ByteBufDataInputPlus.java         |  12 +
 .../net/async/ByteBufDataOutputStreamPlus.java  | 191 +++++++
 .../net/async/InboundHandshakeHandler.java      |  36 +-
 .../cassandra/net/async/NettyFactory.java       |  24 +-
 .../net/async/OutboundConnectionIdentifier.java |  21 +-
 .../net/async/OutboundHandshakeHandler.java     |   9 +-
 .../async/RebufferingByteBufDataInputPlus.java  | 250 +++++++++
 .../apache/cassandra/security/SSLFactory.java   |  49 --
 .../cassandra/service/StorageService.java       |  11 -
 .../cassandra/service/StorageServiceMBean.java  |   3 -
 .../cassandra/streaming/ConnectionHandler.java  | 428 ----------------
 .../streaming/DefaultConnectionFactory.java     | 122 +++--
 .../streaming/StreamConnectionFactory.java      |  11 +-
 .../cassandra/streaming/StreamCoordinator.java  |  22 +-
 .../cassandra/streaming/StreamManager.java      |  24 +-
 .../apache/cassandra/streaming/StreamPlan.java  |   2 +-
 .../cassandra/streaming/StreamReader.java       |  25 +-
 .../streaming/StreamReceiveException.java       |  36 ++
 .../cassandra/streaming/StreamReceiveTask.java  |   1 +
 .../cassandra/streaming/StreamResultFuture.java |  32 +-
 .../cassandra/streaming/StreamSession.java      | 396 ++++++++-------
 .../cassandra/streaming/StreamTransferTask.java |  10 +-
 .../cassandra/streaming/StreamWriter.java       | 115 +++--
 .../streaming/StreamingMessageSender.java       |  34 ++
 .../async/NettyStreamingMessageSender.java      | 508 +++++++++++++++++++
 .../async/StreamCompressionSerializer.java      | 133 +++++
 .../async/StreamingInboundHandler.java          | 268 ++++++++++
 .../cassandra/streaming/async/package-info.java |  71 +++
 .../ByteBufCompressionDataOutputStreamPlus.java |  76 +++
 .../compress/CompressedInputStream.java         | 225 ++++----
 .../compress/CompressedStreamReader.java        |  17 +-
 .../compress/CompressedStreamWriter.java        |  25 +-
 .../compress/StreamCompressionInputStream.java  |  78 +++
 .../streaming/messages/CompleteMessage.java     |  10 +-
 .../streaming/messages/FileMessageHeader.java   |  38 +-
 .../streaming/messages/IncomingFileMessage.java |  30 +-
 .../streaming/messages/KeepAliveMessage.java    |   9 +-
 .../streaming/messages/OutgoingFileMessage.java |  28 +-
 .../streaming/messages/PrepareAckMessage.java   |  57 +++
 .../streaming/messages/PrepareMessage.java      |  93 ----
 .../messages/PrepareSynAckMessage.java          |  80 +++
 .../streaming/messages/PrepareSynMessage.java   |  98 ++++
 .../streaming/messages/ReceivedMessage.java     |  11 +-
 .../streaming/messages/RetryMessage.java        |  71 ---
 .../messages/SessionFailedMessage.java          |  10 +-
 .../streaming/messages/StreamInitMessage.java   |  73 +--
 .../streaming/messages/StreamMessage.java       |  58 +--
 .../tools/BulkLoadConnectionFactory.java        |  32 +-
 .../org/apache/cassandra/tools/NodeProbe.java   |   7 -
 .../cassandra/tools/nodetool/GetTimeout.java    |   2 +-
 .../org/apache/cassandra/utils/UUIDGen.java     |   8 +-
 .../cassandra/streaming/LongStreamingTest.java  |  34 +-
 .../cassandra/cql3/PreparedStatementsTest.java  |   2 +-
 .../util/RewindableDataInputStreamPlusTest.java |   2 +-
 .../net/async/HandshakeHandlersTest.java        |   4 +-
 .../net/async/InboundHandshakeHandlerTest.java  |   8 +-
 .../async/OutboundMessagingConnectionTest.java  |  45 --
 .../RebufferingByteBufDataInputPlusTest.java    | 126 +++++
 .../net/async/TestScheduledFuture.java          |  66 +++
 .../apache/cassandra/service/RemoveTest.java    |   3 +-
 .../streaming/StreamTransferTaskTest.java       |   7 +-
 .../streaming/StreamingTransferTest.java        |  28 -
 .../async/NettyStreamingMessageSenderTest.java  | 202 ++++++++
 .../async/StreamCompressionSerializerTest.java  | 135 +++++
 .../async/StreamingInboundHandlerTest.java      | 168 ++++++
 .../compression/CompressedInputStreamTest.java  |  10 +-
 75 files changed, 3508 insertions(+), 1504 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index f2e643e..a14e390 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 4.0
+ * use netty for streaming (CASSANDRA-12229)
  * Use netty for internode messaging (CASSANDRA-8457)
  * Add bytes repaired/unrepaired to nodetool tablestats (CASSANDRA-13774)
  * Don't delete incremental repair sessions if they still have sstables (CASSANDRA-13758)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/lib/compress-lzf-0.8.4.jar
----------------------------------------------------------------------
diff --git a/lib/compress-lzf-0.8.4.jar b/lib/compress-lzf-0.8.4.jar
deleted file mode 100644
index a712c24..0000000
Binary files a/lib/compress-lzf-0.8.4.jar and /dev/null differ

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/config/Config.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java
index 77d5bf4..537cf39 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -97,12 +97,6 @@ public class Config
 
     public volatile long truncate_request_timeout_in_ms = 60000L;
 
-    /**
-     * @deprecated use {@link #streaming_keep_alive_period_in_secs} instead
-     */
-    @Deprecated
-    public int streaming_socket_timeout_in_ms = 86400000; //24 hours
-
     public Integer streaming_connections_per_host = 1;
     public Integer streaming_keep_alive_period_in_secs = 300; //5 minutes
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 53bac93..302a528 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -2060,21 +2060,6 @@ public class DatabaseDescriptor
         conf.counter_cache_keys_to_save = counterCacheKeysToSave;
     }
 
-    public static void setStreamingSocketTimeout(int value)
-    {
-        conf.streaming_socket_timeout_in_ms = value;
-    }
-
-    /**
-     * @deprecated use {@link #getStreamingKeepAlivePeriod()} instead
-     * @return streaming_socket_timeout_in_ms property
-     */
-    @Deprecated
-    public static int getStreamingSocketTimeout()
-    {
-        return conf.streaming_socket_timeout_in_ms;
-    }
-
     public static int getStreamingKeepAlivePeriod()
     {
         return conf.streaming_keep_alive_period_in_secs;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/exceptions/ChecksumMismatchException.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/exceptions/ChecksumMismatchException.java b/src/java/org/apache/cassandra/exceptions/ChecksumMismatchException.java
new file mode 100644
index 0000000..a76c46c
--- /dev/null
+++ b/src/java/org/apache/cassandra/exceptions/ChecksumMismatchException.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.exceptions;
+
+import java.io.IOException;
+
+public class ChecksumMismatchException extends IOException
+{
+    public ChecksumMismatchException()
+    {
+        super();
+    }
+
+    public ChecksumMismatchException(String s)
+    {
+        super(s);
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java
index 6c1849f..8ac6589 100644
--- a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java
+++ b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java
@@ -145,7 +145,9 @@ public class CompressionMetadata
         this.chunkOffsetsSize = chunkOffsets.size();
     }
 
-    private CompressionMetadata(String filePath, CompressionParams parameters, SafeMemory offsets, long offsetsSize, long dataLength, long compressedLength)
+    // do not call this constructor directly, unless used in testing
+    @VisibleForTesting
+    public CompressionMetadata(String filePath, CompressionParams parameters, Memory offsets, long offsetsSize, long dataLength, long compressedLength)
     {
         this.indexFilePath = filePath;
         this.parameters = parameters;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
index ff47bec..dc56520 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
@@ -208,7 +208,7 @@ public class SSTableLoader implements StreamEventHandler
         for (SSTableReader sstable : sstables)
         {
             sstable.selfRef().release();
-            assert sstable.selfRef().globalCount() == 0;
+            assert sstable.selfRef().globalCount() == 0 : String.format("for sstable = %s, ref count = %d", sstable, sstable.selfRef().globalCount());
         }
     }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java b/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java
index 91b189d..277b359 100644
--- a/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java
+++ b/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java
@@ -20,9 +20,12 @@ package org.apache.cassandra.io.util;
 import java.io.Closeable;
 import java.io.File;
 import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.util.zip.CheckedInputStream;
 import java.util.zip.Checksum;
 
+import com.google.common.annotations.VisibleForTesting;
+
 import org.apache.cassandra.io.sstable.Component;
 import org.apache.cassandra.io.sstable.Descriptor;
 import org.apache.cassandra.utils.ChecksumType;
@@ -57,6 +60,15 @@ public class DataIntegrityMetadata
             chunkSize = reader.readInt();
         }
 
+        @VisibleForTesting
+        protected ChecksumValidator(ChecksumType checksumType, RandomAccessReader reader, int chunkSize)
+        {
+            this.checksumType = checksumType;
+            this.reader = reader;
+            this.dataFilename = null;
+            this.chunkSize = chunkSize;
+        }
+
         public void seek(long offset)
         {
             long start = chunkStart(offset);
@@ -77,6 +89,20 @@ public class DataIntegrityMetadata
                 throw new IOException("Corrupted File : " + dataFilename);
         }
 
+        /**
+         * validates the checksum with the bytes from the specified buffer.
+         *
+         * Upon return, the buffer's position will
+         * be updated to its limit; its limit will not have been changed.
+         */
+        public void validate(ByteBuffer buffer) throws IOException
+        {
+            int current = (int) checksumType.of(buffer);
+            int actual = reader.readInt();
+            if (current != actual)
+                throw new IOException("Corrupted File : " + dataFilename);
+        }
+
         public void close()
         {
             reader.close();

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
deleted file mode 100644
index 8db5fcb..0000000
--- a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java
+++ /dev/null
@@ -1,104 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.cassandra.net;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.net.Socket;
-import java.util.Set;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.cassandra.io.util.DataInputPlus;
-import org.apache.cassandra.io.util.DataInputPlus.DataInputStreamPlus;
-import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.streaming.StreamResultFuture;
-import org.apache.cassandra.streaming.messages.StreamInitMessage;
-import org.apache.cassandra.streaming.messages.StreamMessage;
-
-/**
- * Thread to consume stream init messages.
- */
-public class IncomingStreamingConnection extends Thread implements Closeable
-{
-    private static final Logger logger = LoggerFactory.getLogger(IncomingStreamingConnection.class);
-
-    private final int version;
-    public final Socket socket;
-    private final Set<Closeable> group;
-
-    public IncomingStreamingConnection(int version, Socket socket, Set<Closeable> group)
-    {
-        super("STREAM-INIT-" + socket.getRemoteSocketAddress());
-        this.version = version;
-        this.socket = socket;
-        this.group = group;
-    }
-
-    @Override
-    @SuppressWarnings("resource") // Not closing constructed DataInputPlus's as the stream needs to remain open.
-    public void run()
-    {
-        try
-        {
-            // streaming connections are per-session and have a fixed version.
-            // we can't do anything with a wrong-version stream connection, so drop it.
-            if (version != StreamMessage.CURRENT_VERSION)
-                throw new IOException(String.format("Received stream using protocol version %d (my version %d). Terminating connection", version, StreamMessage.CURRENT_VERSION));
-
-            DataInputPlus input = new DataInputStreamPlus(socket.getInputStream());
-            StreamInitMessage init = StreamInitMessage.serializer.deserialize(input, version);
-
-            //Set SO_TIMEOUT on follower side
-            if (!init.isForOutgoing)
-                socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
-
-            // The initiator makes two connections, one for incoming and one for outgoing.
-            // The receiving side distinguish two connections by looking at StreamInitMessage#isForOutgoing.
-            // Note: we cannot use the same socket for incoming and outgoing streams because we want to
-            // parallelize said streams and the socket is blocking, so we might deadlock.
-            StreamResultFuture.initReceivingSide(init.sessionIndex, init.planId, init.streamOperation, init.from, this, init.isForOutgoing, version, init.keepSSTableLevel, init.pendingRepair, init.previewKind);
-        }
-        catch (Throwable t)
-        {
-            logger.error("Error while reading from socket from {}.", socket.getRemoteSocketAddress(), t);
-            close();
-        }
-    }
-
-    @Override
-    public void close()
-    {
-        try
-        {
-            if (!socket.isClosed())
-            {
-                socket.close();
-            }
-        }
-        catch (IOException e)
-        {
-            logger.debug("Error closing socket", e);
-        }
-        finally
-        {
-            group.remove(this);
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/net/async/ByteBufDataInputPlus.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/async/ByteBufDataInputPlus.java b/src/java/org/apache/cassandra/net/async/ByteBufDataInputPlus.java
index f9fa07a..23e532c 100644
--- a/src/java/org/apache/cassandra/net/async/ByteBufDataInputPlus.java
+++ b/src/java/org/apache/cassandra/net/async/ByteBufDataInputPlus.java
@@ -24,8 +24,20 @@ import org.apache.cassandra.io.util.DataInputPlus;
 
 public class ByteBufDataInputPlus extends ByteBufInputStream implements DataInputPlus
 {
+    /**
+     * The parent class does not expose the buffer to derived classes, so we need
+     * to stash a reference here so it can be exposed via {@link #buffer()}.
+     */
+    private final ByteBuf buf;
+
     public ByteBufDataInputPlus(ByteBuf buffer)
     {
         super(buffer);
+        this.buf = buffer;
+    }
+
+    public ByteBuf buffer()
+    {
+        return buf;
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/net/async/ByteBufDataOutputStreamPlus.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/async/ByteBufDataOutputStreamPlus.java b/src/java/org/apache/cassandra/net/async/ByteBufDataOutputStreamPlus.java
new file mode 100644
index 0000000..3a544e4
--- /dev/null
+++ b/src/java/org/apache/cassandra/net/async/ByteBufDataOutputStreamPlus.java
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.net.async;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.WritableByteChannel;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.util.concurrent.Uninterruptibles;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.util.concurrent.Future;
+import org.apache.cassandra.io.util.BufferedDataOutputStreamPlus;
+import org.apache.cassandra.io.util.DataOutputStreamPlus;
+import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.streaming.StreamSession;
+
+/**
+ * A {@link DataOutputStreamPlus} that writes to a {@link ByteBuf}. The novelty here is that all writes
+ * actually get written in to a {@link ByteBuffer} that shares a backing buffer with a {@link ByteBuf}.
+ * The trick to do that is allocate the ByteBuf, get a ByteBuffer from it by calling {@link ByteBuf#nioBuffer()},
+ * and passing that to the super class as {@link #buffer}. When the {@link #buffer} is full or {@link #doFlush(int)}
+ * is invoked, the {@link #currentBuf} is published to the netty channel.
+ */
+public class ByteBufDataOutputStreamPlus extends BufferedDataOutputStreamPlus
+{
+    private final StreamSession session;
+    private final Channel channel;
+    private final int bufferSize;
+
+    /**
+     * Tracks how many bytes we've written to the netty channel. This more or less follows the channel's
+     * high/low water marks and ultimately the 'writablility' status of the channel. Unfortunately there's
+     * no notification mechanism that can poke a producer to let it know when the channel becomes writable
+     * (after it was unwritable); hence, the use of a {@link Semaphore}.
+     */
+    private final Semaphore channelRateLimiter;
+
+    /**
+     * This *must* be the owning {@link ByteBuf} for the {@link BufferedDataOutputStreamPlus#buffer}
+     */
+    private ByteBuf currentBuf;
+
+    private ByteBufDataOutputStreamPlus(StreamSession session, Channel channel, ByteBuf buffer, int bufferSize)
+    {
+        super(buffer.nioBuffer(0, bufferSize));
+        this.session = session;
+        this.channel = channel;
+        this.currentBuf = buffer;
+        this.bufferSize = bufferSize;
+
+        channelRateLimiter = new Semaphore(channel.config().getWriteBufferHighWaterMark(), true);
+    }
+
+    @Override
+    protected WritableByteChannel newDefaultChannel()
+    {
+        return new WritableByteChannel()
+        {
+            @Override
+            public int write(ByteBuffer src) throws IOException
+            {
+                assert src == buffer;
+                int size = src.position();
+                doFlush(size);
+                return size;
+            }
+
+            @Override
+            public boolean isOpen()
+            {
+                return channel.isOpen();
+            }
+
+            @Override
+            public void close()
+            {   }
+        };
+    }
+
+    public static ByteBufDataOutputStreamPlus create(StreamSession session, Channel channel, int bufferSize)
+    {
+        ByteBuf buf = channel.alloc().directBuffer(bufferSize, bufferSize);
+        return new ByteBufDataOutputStreamPlus(session, channel, buf, bufferSize);
+    }
+
+    /**
+     * Writes the incoming buffer directly to the backing {@link #channel}, without copying to the intermediate {@link #buffer}.
+     */
+    public ChannelFuture writeToChannel(ByteBuf buf) throws IOException
+    {
+        doFlush(buffer.position());
+
+        int byteCount = buf.readableBytes();
+        if (!Uninterruptibles.tryAcquireUninterruptibly(channelRateLimiter, byteCount, 5, TimeUnit.MINUTES))
+            throw new IOException("outbound channel was not writable");
+
+        // the (possibly naive) assumption that we should always flush after each incoming buf
+        ChannelFuture channelFuture = channel.writeAndFlush(buf);
+        channelFuture.addListener(future -> handleBuffer(future, byteCount));
+        return channelFuture;
+    }
+
+    /**
+     * Writes the incoming buffer directly to the backing {@link #channel}, without copying to the intermediate {@link #buffer}.
+     * The incoming buffer will be automatically released when the netty channel invokes the listeners of success/failure to
+     * send the buffer.
+     */
+    public ChannelFuture writeToChannel(ByteBuffer buffer) throws IOException
+    {
+        ChannelFuture channelFuture = writeToChannel(Unpooled.wrappedBuffer(buffer));
+        channelFuture.addListener(future -> FileUtils.clean(buffer));
+        return channelFuture;
+    }
+
+    @Override
+    protected void doFlush(int count) throws IOException
+    {
+        // flush the current backing write buffer only if there's any pending data
+        if (buffer.position() > 0 && channel.isOpen())
+        {
+            int byteCount = buffer.position();
+            currentBuf.writerIndex(byteCount);
+
+            if (!Uninterruptibles.tryAcquireUninterruptibly(channelRateLimiter, byteCount, 2, TimeUnit.MINUTES))
+                throw new IOException("outbound channel was not writable");
+
+            channel.writeAndFlush(currentBuf).addListener(future -> handleBuffer(future, byteCount));
+            currentBuf = channel.alloc().directBuffer(bufferSize, bufferSize);
+            buffer = currentBuf.nioBuffer(0, bufferSize);
+        }
+    }
+
+    /**
+     * Handles the result of publishing a buffer to the channel.
+     *
+     * Note: this will be executed on the event loop.
+     */
+    private void handleBuffer(Future<? super Void> future, int bytesWritten)
+    {
+        channelRateLimiter.release(bytesWritten);
+
+        if (!future.isSuccess() && channel.isOpen())
+            session.onError(future.cause());
+    }
+
+    public ByteBufAllocator getAllocator()
+    {
+        return channel.alloc();
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * Flush any last buffered (if the channel is open), and release any buffers. *Not* responsible for closing
+     * the netty channel as we might use it again for transferring more files.
+     *
+     * Note: should be called on the producer thread, not the netty event loop.
+     */
+    @Override
+    public void close() throws IOException
+    {
+        doFlush(0);
+        if (currentBuf.refCnt() > 0)
+            currentBuf.release();
+        currentBuf = null;
+        buffer = null;
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/net/async/InboundHandshakeHandler.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/async/InboundHandshakeHandler.java b/src/java/org/apache/cassandra/net/async/InboundHandshakeHandler.java
index 5ea03dc..7a8303c 100644
--- a/src/java/org/apache/cassandra/net/async/InboundHandshakeHandler.java
+++ b/src/java/org/apache/cassandra/net/async/InboundHandshakeHandler.java
@@ -14,6 +14,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.netty.buffer.ByteBuf;
+import io.netty.channel.AdaptiveRecvByteBufAllocator;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPipeline;
@@ -25,6 +26,8 @@ import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.net.async.HandshakeProtocol.FirstHandshakeMessage;
 import org.apache.cassandra.net.async.HandshakeProtocol.SecondHandshakeMessage;
 import org.apache.cassandra.net.async.HandshakeProtocol.ThirdHandshakeMessage;
+import org.apache.cassandra.streaming.async.StreamingInboundHandler;
+import org.apache.cassandra.streaming.messages.StreamMessage;
 
 /**
  * 'Server'-side component that negotiates the internode handshake when establishing a new connection.
@@ -36,13 +39,13 @@ class InboundHandshakeHandler extends ByteToMessageDecoder
 {
     private static final Logger logger = LoggerFactory.getLogger(NettyFactory.class);
 
-    enum State { START, AWAITING_HANDSHAKE_BEGIN, AWAIT_STREAM_START_RESPONSE, AWAIT_MESSAGING_START_RESPONSE, MESSAGING_HANDSHAKE_COMPLETE, HANDSHAKE_FAIL }
+    enum State { START, AWAITING_HANDSHAKE_BEGIN, AWAIT_MESSAGING_START_RESPONSE, HANDSHAKE_COMPLETE, HANDSHAKE_FAIL }
 
     private State state;
 
     private final IInternodeAuthenticator authenticator;
-    private boolean hasAuthenticated;
 
+    private boolean hasAuthenticated;
     /**
      * The peer's declared messaging version.
      */
@@ -160,9 +163,16 @@ class InboundHandshakeHandler extends ByteToMessageDecoder
 
         if (msg.mode == NettyFactory.Mode.STREAMING)
         {
-            // TODO fill in once streaming is moved to netty
-            ctx.close();
-            return State.AWAIT_STREAM_START_RESPONSE;
+            // streaming connections are per-session and have a fixed version.  we can't do anything with a wrong-version stream connection, so drop it.
+            if (version != StreamMessage.CURRENT_VERSION)
+            {
+                logger.warn("Received stream using protocol version %d (my version %d). Terminating connection", version, MessagingService.current_version);
+                ctx.close();
+                return State.HANDSHAKE_FAIL;
+            }
+
+            setupStreamingPipeline(ctx, version);
+            return State.HANDSHAKE_COMPLETE;
         }
         else
         {
@@ -195,6 +205,18 @@ class InboundHandshakeHandler extends ByteToMessageDecoder
         }
     }
 
+    private void setupStreamingPipeline(ChannelHandlerContext ctx, int protocolVersion)
+    {
+        ChannelPipeline pipeline = ctx.pipeline();
+        InetSocketAddress address = (InetSocketAddress) ctx.channel().remoteAddress();
+        pipeline.addLast(NettyFactory.instance.streamingGroup, "streamInbound", new StreamingInboundHandler(address, protocolVersion, null));
+        pipeline.remove(this);
+
+        // pass a custom recv ByteBuf allocator to the channel. the default recv ByteBuf size is 1k, but in streaming we're
+        // dealing with large bulk blocks of data, let's default to larger sizes
+        ctx.channel().config().setRecvByteBufAllocator(new AdaptiveRecvByteBufAllocator(1 << 8, 1 << 13, 1 << 16));
+    }
+
     /**
      * Handles the third (and last) message in the internode messaging handshake protocol. Grabs the protocol version and
      * IP addr the peer wants to use.
@@ -227,7 +249,7 @@ class InboundHandshakeHandler extends ByteToMessageDecoder
         logger.trace("Set version for {} to {} (will use {})", from, maxVersion, MessagingService.instance().getVersion(from));
 
         setupMessagingPipeline(ctx.pipeline(), from, compressed, version);
-        return State.MESSAGING_HANDSHAKE_COMPLETE;
+        return State.HANDSHAKE_COMPLETE;
     }
 
     @VisibleForTesting
@@ -245,7 +267,7 @@ class InboundHandshakeHandler extends ByteToMessageDecoder
     {
         // we're not really racing on the handshakeTimeout as we're in the event loop,
         // but, hey, defensive programming is beautiful thing!
-        if (state == State.MESSAGING_HANDSHAKE_COMPLETE || (handshakeTimeout != null && handshakeTimeout.isCancelled()))
+        if (state == State.HANDSHAKE_COMPLETE || (handshakeTimeout != null && handshakeTimeout.isCancelled()))
             return;
 
         state = State.HANDSHAKE_FAIL;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/net/async/NettyFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/async/NettyFactory.java b/src/java/org/apache/cassandra/net/async/NettyFactory.java
index 13d8810..762c39b 100644
--- a/src/java/org/apache/cassandra/net/async/NettyFactory.java
+++ b/src/java/org/apache/cassandra/net/async/NettyFactory.java
@@ -40,6 +40,7 @@ import io.netty.util.concurrent.DefaultThreadFactory;
 import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 import io.netty.util.internal.logging.Slf4JLoggerFactory;
+
 import net.jpountz.lz4.LZ4Factory;
 import net.jpountz.xxhash.XXHashFactory;
 import org.apache.cassandra.auth.IInternodeAuthenticator;
@@ -69,12 +70,18 @@ public final class NettyFactory
 
     private static final int LZ4_HASH_SEED = 0x9747b28c;
 
+    /**
+     * Default seed value for xxhash.
+     */
+    public static final int XXHASH_DEFAULT_SEED = 0x9747b28c;
+
     public enum Mode { MESSAGING, STREAMING }
 
     private static final String SSL_CHANNEL_HANDLER_NAME = "ssl";
-    static final String INBOUND_COMPRESSOR_HANDLER_NAME = "inboundCompressor";
-    static final String OUTBOUND_COMPRESSOR_HANDLER_NAME = "outboundCompressor";
-    private static final String HANDSHAKE_HANDLER_NAME = "handshakeHandler";
+    public static final String INBOUND_COMPRESSOR_HANDLER_NAME = "inboundCompressor";
+    public static final String OUTBOUND_COMPRESSOR_HANDLER_NAME = "outboundCompressor";
+    public static final String HANDSHAKE_HANDLER_NAME = "handshakeHandler";
+    public static final String INBOUND_STREAM_HANDLER_NAME = "inboundStreamHandler";
 
     /** a useful addition for debugging; simply set to true to get more data in your logs */
     private static final boolean WIRETRACE = false;
@@ -113,6 +120,7 @@ public final class NettyFactory
 
     private final EventLoopGroup inboundGroup;
     private final EventLoopGroup outboundGroup;
+    public final EventLoopGroup streamingGroup;
 
     /**
      * Constructor that allows modifying the {@link NettyFactory#useEpoll} for testing purposes. Otherwise, use the
@@ -126,6 +134,7 @@ public final class NettyFactory
                                         "MessagingService-NettyAcceptor-Threads", false);
         inboundGroup = getEventLoopGroup(useEpoll, FBUtilities.getAvailableProcessors(), "MessagingService-NettyInbound-Threads", false);
         outboundGroup = getEventLoopGroup(useEpoll, FBUtilities.getAvailableProcessors(), "MessagingService-NettyOutbound-Threads", true);
+        streamingGroup = getEventLoopGroup(useEpoll, FBUtilities.getAvailableProcessors(), "Streaming-Netty-Threads", false);
     }
 
     /**
@@ -257,7 +266,8 @@ public final class NettyFactory
                 SslContext sslContext = SSLFactory.getSslContext(encryptionOptions, true, true);
                 SslHandler sslHandler = sslContext.newHandler(channel.alloc());
                 logger.trace("creating inbound netty SslContext: context={}, engine={}", sslContext.getClass().getName(), sslHandler.engine().getClass().getName());
-                pipeline.addFirst(SSL_CHANNEL_HANDLER_NAME, sslHandler);            }
+                pipeline.addFirst(SSL_CHANNEL_HANDLER_NAME, sslHandler);
+            }
 
             if (WIRETRACE)
                 pipeline.addLast("logger", new LoggingHandler(LogLevel.INFO));
@@ -279,13 +289,14 @@ public final class NettyFactory
      * Create the {@link Bootstrap} for connecting to a remote peer. This method does <b>not</b> attempt to connect to the peer,
      * and thus does not block.
      */
+    @VisibleForTesting
     public Bootstrap createOutboundBootstrap(OutboundConnectionParams params)
     {
         logger.debug("creating outbound bootstrap to peer {}, compression: {}, encryption: {}, coalesce: {}", params.connectionId.connectionAddress(),
                      params.compress, encryptionLogStatement(params.encryptionOptions),
                      params.coalescingStrategy.isPresent() ? params.coalescingStrategy.get() : CoalescingStrategies.Strategy.DISABLED);
-        Class<? extends Channel>  transport = useEpoll ? EpollSocketChannel.class : NioSocketChannel.class;
-        Bootstrap bootstrap = new Bootstrap().group(outboundGroup)
+        Class<? extends Channel> transport = useEpoll ? EpollSocketChannel.class : NioSocketChannel.class;
+        Bootstrap bootstrap = new Bootstrap().group(params.mode == Mode.MESSAGING ? outboundGroup : streamingGroup)
                               .channel(transport)
                               .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 2000)
                               .option(ChannelOption.SO_KEEPALIVE, true)
@@ -349,6 +360,7 @@ public final class NettyFactory
         acceptGroup.shutdownGracefully();
         outboundGroup.shutdownGracefully();
         inboundGroup.shutdownGracefully();
+        streamingGroup.shutdownGracefully();
     }
 
     static Lz4FrameEncoder createLz4Encoder(int protocolVersion)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/net/async/OutboundConnectionIdentifier.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/async/OutboundConnectionIdentifier.java b/src/java/org/apache/cassandra/net/async/OutboundConnectionIdentifier.java
index 24dc5ff..c834bd4 100644
--- a/src/java/org/apache/cassandra/net/async/OutboundConnectionIdentifier.java
+++ b/src/java/org/apache/cassandra/net/async/OutboundConnectionIdentifier.java
@@ -32,7 +32,7 @@ public class OutboundConnectionIdentifier
 {
     enum ConnectionType
     {
-        GOSSIP, LARGE_MESSAGE, SMALL_MESSAGE
+        GOSSIP, LARGE_MESSAGE, SMALL_MESSAGE, STREAM
     }
 
     /**
@@ -99,6 +99,15 @@ public class OutboundConnectionIdentifier
     }
 
     /**
+     * Creates an identifier for a gossip connection and using the remote "identifying" address as its connection
+     * address.
+     */
+    public static OutboundConnectionIdentifier stream(InetSocketAddress localAddr, InetSocketAddress remoteAddr)
+    {
+        return new OutboundConnectionIdentifier(localAddr, remoteAddr, ConnectionType.STREAM);
+    }
+
+    /**
      * Returns a newly created connection identifier to the same remote that this identifier, but using the provided
      * address as connection address.
      *
@@ -106,7 +115,7 @@ public class OutboundConnectionIdentifier
      * @return a newly created connection identifier that differs from this one only by using {@code remoteConnectionAddr}
      * as connection address to the remote.
      */
-    OutboundConnectionIdentifier withNewConnectionAddress(InetSocketAddress remoteConnectionAddr)
+    public OutboundConnectionIdentifier withNewConnectionAddress(InetSocketAddress remoteConnectionAddr)
     {
         return new OutboundConnectionIdentifier(localAddr, remoteAddr, remoteConnectionAddr, connectionType);
     }
@@ -114,7 +123,7 @@ public class OutboundConnectionIdentifier
     /**
      * The local node address.
      */
-    InetAddress local()
+    public InetAddress local()
     {
         return localAddr.getAddress();
     }
@@ -122,7 +131,7 @@ public class OutboundConnectionIdentifier
     /**
      * The remote node identifying address (the one to use for anything else than connecting to the node).
      */
-    InetSocketAddress remoteAddress()
+    public InetSocketAddress remoteAddress()
     {
         return remoteAddr;
     }
@@ -130,7 +139,7 @@ public class OutboundConnectionIdentifier
     /**
      * The remote node identifying address (the one to use for anything else than connecting to the node).
      */
-    InetAddress remote()
+    public  InetAddress remote()
     {
         return remoteAddr.getAddress();
     }
@@ -138,7 +147,7 @@ public class OutboundConnectionIdentifier
     /**
      * The remote node connection address (the one to use to actually connect to the remote, and only that).
      */
-    InetSocketAddress connectionAddress()
+    public InetSocketAddress connectionAddress()
     {
         return remoteConnectionAddr;
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/net/async/OutboundHandshakeHandler.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/async/OutboundHandshakeHandler.java b/src/java/org/apache/cassandra/net/async/OutboundHandshakeHandler.java
index 703549a..c555bed 100644
--- a/src/java/org/apache/cassandra/net/async/OutboundHandshakeHandler.java
+++ b/src/java/org/apache/cassandra/net/async/OutboundHandshakeHandler.java
@@ -36,6 +36,7 @@ import io.netty.channel.ChannelPipeline;
 import io.netty.handler.codec.ByteToMessageDecoder;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.concurrent.Future;
+
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.net.async.HandshakeProtocol.FirstHandshakeMessage;
@@ -95,7 +96,9 @@ public class OutboundHandshakeHandler extends ByteToMessageDecoder
     /**
      * {@inheritDoc}
      *
-     * Invoked when the channel is made active, and sends out the {@link FirstHandshakeMessage}
+     * Invoked when the channel is made active, and sends out the {@link FirstHandshakeMessage}.
+     * In the case of streaming, we do not require a full bi-directional handshake; the initial message,
+     * containing the streaming protocol version, is all that is required.
      */
     @Override
     public void channelActive(final ChannelHandlerContext ctx) throws Exception
@@ -103,6 +106,10 @@ public class OutboundHandshakeHandler extends ByteToMessageDecoder
         FirstHandshakeMessage msg = new FirstHandshakeMessage(messagingVersion, mode, params.compress);
         logger.trace("starting handshake with peer {}, msg = {}", connectionId.connectionAddress(), msg);
         ctx.writeAndFlush(msg.encode(ctx.alloc())).addListener(future -> firstHandshakeMessageListener(future, ctx));
+
+        if (mode == NettyFactory.Mode.STREAMING)
+            ctx.pipeline().remove(this);
+
         ctx.fireChannelActive();
     }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/net/async/RebufferingByteBufDataInputPlus.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/async/RebufferingByteBufDataInputPlus.java b/src/java/org/apache/cassandra/net/async/RebufferingByteBufDataInputPlus.java
new file mode 100644
index 0000000..580bc03
--- /dev/null
+++ b/src/java/org/apache/cassandra/net/async/RebufferingByteBufDataInputPlus.java
@@ -0,0 +1,250 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.net.async;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelConfig;
+import io.netty.util.ReferenceCountUtil;
+import org.apache.cassandra.io.util.RebufferingInputStream;
+
+public class RebufferingByteBufDataInputPlus extends RebufferingInputStream implements ReadableByteChannel
+{
+    /**
+     * The parent, or owning, buffer of the current buffer being read from ({@link super#buffer}).
+     */
+    private ByteBuf currentBuf;
+
+    private final BlockingQueue<ByteBuf> queue;
+
+    /**
+     * The count of live bytes in all {@link ByteBuf}s held by this instance.
+     */
+    private final AtomicInteger queuedByteCount;
+
+    private final int lowWaterMark;
+    private final int highWaterMark;
+    private final ChannelConfig channelConfig;
+
+    private volatile boolean closed;
+
+    public RebufferingByteBufDataInputPlus(int lowWaterMark, int highWaterMark, ChannelConfig channelConfig)
+    {
+        super(Unpooled.EMPTY_BUFFER.nioBuffer());
+
+        if (lowWaterMark > highWaterMark)
+            throw new IllegalArgumentException(String.format("low water mark is greater than high water mark: %d vs %d", lowWaterMark, highWaterMark));
+
+        currentBuf = Unpooled.EMPTY_BUFFER;
+        this.lowWaterMark = lowWaterMark;
+        this.highWaterMark = highWaterMark;
+        this.channelConfig = channelConfig;
+        queue = new LinkedBlockingQueue<>();
+        queuedByteCount = new AtomicInteger();
+    }
+
+    /**
+     * Append a {@link ByteBuf} to the end of the einternal queue.
+     *
+     * Note: it's expected this method is invoked on the netty event loop.
+     */
+    public void append(ByteBuf buf) throws IllegalStateException
+    {
+        assert buf != null : "buffer cannot be null";
+
+        if (closed)
+        {
+            ReferenceCountUtil.release(buf);
+            throw new IllegalStateException("stream is already closed, so cannot add another buffer");
+        }
+
+        // this slightly undercounts the live count as it doesn't include the currentBuf's size.
+        // that's ok as the worst we'll do is allow another buffer in and add it to the queue,
+        // and that point we'll disable auto-read. this is a tradeoff versus making some other member field
+        // atomic or volatile.
+        int queuedCount = queuedByteCount.addAndGet(buf.readableBytes());
+        if (channelConfig.isAutoRead() && queuedCount > highWaterMark)
+            channelConfig.setAutoRead(false);
+
+        queue.add(buf);
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * Release open buffers and poll the {@link #queue} for more data.
+     * <p>
+     * This is best, and more or less expected, to be invoked on a consuming thread (not the event loop)
+     * becasue if we block on the queue we can't fill it on the event loop (as that's where the buffers are coming from).
+     */
+    @Override
+    protected void reBuffer() throws IOException
+    {
+        currentBuf.release();
+        buffer = null;
+        currentBuf = null;
+
+        // possibly re-enable auto-read, *before* blocking on the queue, because if we block on the queue
+        // without enabling auto-read we'll block forever :(
+        if (!channelConfig.isAutoRead() && queuedByteCount.get() < lowWaterMark)
+            channelConfig.setAutoRead(true);
+
+        try
+        {
+            currentBuf = queue.take();
+            int bytes;
+            // if we get an explicitly empty buffer, we treat that as an indicator that the input is closed
+            if (currentBuf == null || (bytes = currentBuf.readableBytes()) == 0)
+            {
+                releaseResources();
+                throw new EOFException();
+            }
+
+            buffer = currentBuf.nioBuffer(currentBuf.readerIndex(), bytes);
+            assert buffer.remaining() == bytes;
+            queuedByteCount.addAndGet(-bytes);
+            return;
+        }
+        catch (InterruptedException ie)
+        {
+            // nop - ignore
+        }
+    }
+
+    @Override
+    public int read(ByteBuffer dst) throws IOException
+    {
+        int readLength = dst.remaining();
+        int remaining = readLength;
+
+        while (remaining > 0)
+        {
+            if (closed)
+                throw new EOFException();
+
+            if (!buffer.hasRemaining())
+                reBuffer();
+            int copyLength = Math.min(remaining, buffer.remaining());
+
+            int originalLimit = buffer.limit();
+            buffer.limit(buffer.position() + copyLength);
+            dst.put(buffer);
+            buffer.limit(originalLimit);
+            remaining -= copyLength;
+        }
+
+        return readLength;
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * As long as this method is invoked on the consuming thread the returned value will be accurate.
+     */
+    @Override
+    public int available() throws EOFException
+    {
+        if (closed)
+            throw new EOFException();
+
+       final  int availableBytes = queuedByteCount.get() + (buffer != null ? buffer.remaining() : 0);
+
+        if (!channelConfig.isAutoRead() && availableBytes < lowWaterMark)
+            channelConfig.setAutoRead(true);
+
+        return availableBytes;
+    }
+
+    @Override
+    public boolean isOpen()
+    {
+        return !closed;
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * Note: This should invoked on the consuming thread.
+     */
+    @Override
+    public void close()
+    {
+        closed = true;
+        releaseResources();
+    }
+
+    private void releaseResources()
+    {
+        if (currentBuf != null)
+        {
+            if (currentBuf.refCnt() > 0)
+                currentBuf.release(currentBuf.refCnt());
+            currentBuf = null;
+            buffer = null;
+        }
+
+        ByteBuf buf;
+        while ((buf = queue.poll()) != null && buf.refCnt() > 0)
+            buf.release(buf.refCnt());
+    }
+
+    /**
+     * Mark this stream as closed, but do not release any of the resources.
+     *
+     * Note: this is best to be called from the producer thread.
+     */
+    public void markClose()
+    {
+        if (!closed)
+        {
+            closed = true;
+            queue.add(Unpooled.EMPTY_BUFFER);
+        }
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * Note: this is best to be called from the consumer thread.
+     */
+    @Override
+    public String toString()
+    {
+        return new StringBuilder(128).append("RebufferingByteBufDataInputPlus: currentBuf = ").append(currentBuf)
+                                  .append(" (super.buffer = ").append(buffer).append(')')
+                                  .append(", queuedByteCount = ").append(queuedByteCount)
+                                  .append(", queue buffers = ").append(queue)
+                                  .append(", closed = ").append(closed)
+                                  .toString();
+    }
+
+    public ByteBufAllocator getAllocator()
+    {
+        return channelConfig.getAllocator();
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/security/SSLFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/security/SSLFactory.java b/src/java/org/apache/cassandra/security/SSLFactory.java
index 3c1293f..a931f5f 100644
--- a/src/java/org/apache/cassandra/security/SSLFactory.java
+++ b/src/java/org/apache/cassandra/security/SSLFactory.java
@@ -20,7 +20,6 @@ package org.apache.cassandra.security;
 
 import java.io.IOException;
 import java.io.InputStream;
-import java.net.InetAddress;
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.security.KeyStore;
@@ -32,7 +31,6 @@ import java.util.List;
 import java.util.concurrent.atomic.AtomicReference;
 import javax.net.ssl.KeyManagerFactory;
 import javax.net.ssl.SSLContext;
-import javax.net.ssl.SSLParameters;
 import javax.net.ssl.SSLSocket;
 import javax.net.ssl.TrustManager;
 import javax.net.ssl.TrustManagerFactory;
@@ -79,53 +77,6 @@ public final class SSLFactory
      */
     private static final AtomicReference<SslContext> serverSslContext = new AtomicReference<>();
 
-    /** Create a socket and connect */
-    public static SSLSocket getSocket(EncryptionOptions options, InetAddress address, int port, InetAddress localAddress, int localPort) throws IOException
-    {
-        SSLContext ctx = createSSLContext(options, true);
-        SSLSocket socket = (SSLSocket) ctx.getSocketFactory().createSocket(address, port, localAddress, localPort);
-        try
-        {
-            prepareSocket(socket, options);
-            return socket;
-        }
-        catch (IllegalArgumentException e)
-        {
-            socket.close();
-            throw e;
-        }
-    }
-
-    /** Create a socket and connect, using any local address */
-    public static SSLSocket getSocket(EncryptionOptions options, InetAddress address, int port) throws IOException
-    {
-        SSLContext ctx = createSSLContext(options, true);
-        SSLSocket socket = (SSLSocket) ctx.getSocketFactory().createSocket(address, port);
-        try
-        {
-            prepareSocket(socket, options);
-            return socket;
-        }
-        catch (IllegalArgumentException e)
-        {
-            socket.close();
-            throw e;
-        }
-    }
-
-    /** Sets relevant socket options specified in encryption settings */
-    private static void prepareSocket(SSLSocket socket, EncryptionOptions options)
-    {
-        String[] suites = filterCipherSuites(socket.getSupportedCipherSuites(), options.cipher_suites);
-        if(options.require_endpoint_verification)
-        {
-            SSLParameters sslParameters = socket.getSSLParameters();
-            sslParameters.setEndpointIdentificationAlgorithm("HTTPS");
-            socket.setSSLParameters(sslParameters);
-        }
-        socket.setEnabledCipherSuites(suites);
-    }
-
     /**
      * Create a JSSE {@link SSLContext}.
      */

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/service/StorageService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index af59733..bab161a 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -1313,17 +1313,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         return DatabaseDescriptor.getTruncateRpcTimeout();
     }
 
-    public void setStreamingSocketTimeout(int value)
-    {
-        DatabaseDescriptor.setStreamingSocketTimeout(value);
-        logger.info("set streaming socket timeout to {} ms", value);
-    }
-
-    public int getStreamingSocketTimeout()
-    {
-        return DatabaseDescriptor.getStreamingSocketTimeout();
-    }
-
     public void setStreamThroughputMbPerSec(int value)
     {
         DatabaseDescriptor.setStreamThroughputOutboundMegabitsPerSec(value);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
index 36c43fd..46b7253 100644
--- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java
+++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
@@ -502,9 +502,6 @@ public interface StorageServiceMBean extends NotificationEmitter
     public void setTruncateRpcTimeout(long value);
     public long getTruncateRpcTimeout();
 
-    public void setStreamingSocketTimeout(int value);
-    public int getStreamingSocketTimeout();
-
     public void setStreamThroughputMbPerSec(int value);
     public int getStreamThroughputMbPerSec();
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
deleted file mode 100644
index 5f734c9..0000000
--- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java
+++ /dev/null
@@ -1,428 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.cassandra.streaming;
-
-import java.io.BufferedOutputStream;
-import java.io.IOException;
-import java.net.Socket;
-import java.net.SocketException;
-import java.nio.ByteBuffer;
-import java.nio.channels.Channels;
-import java.nio.channels.ReadableByteChannel;
-import java.nio.channels.WritableByteChannel;
-import java.util.Collection;
-import java.util.Comparator;
-import java.util.concurrent.PriorityBlockingQueue;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
-
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.SettableFuture;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import io.netty.util.concurrent.FastThreadLocalThread;
-import org.apache.cassandra.io.util.DataOutputStreamPlus;
-import org.apache.cassandra.io.util.BufferedDataOutputStreamPlus;
-import org.apache.cassandra.io.util.WrappedDataOutputStreamPlus;
-import org.apache.cassandra.net.IncomingStreamingConnection;
-import org.apache.cassandra.streaming.messages.StreamInitMessage;
-import org.apache.cassandra.streaming.messages.StreamMessage;
-import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.JVMStabilityInspector;
-
-/**
- * ConnectionHandler manages incoming/outgoing message exchange for the {@link StreamSession}.
- *
- * <p>
- * Internally, ConnectionHandler manages thread to receive incoming {@link StreamMessage} and thread to
- * send outgoing message. Messages are encoded/decoded on those thread and handed to
- * {@link StreamSession#messageReceived(org.apache.cassandra.streaming.messages.StreamMessage)}.
- */
-public class ConnectionHandler
-{
-    private static final Logger logger = LoggerFactory.getLogger(ConnectionHandler.class);
-
-    private final StreamSession session;
-
-    private IncomingMessageHandler incoming;
-    private OutgoingMessageHandler outgoing;
-    private final boolean isPreview;
-
-    ConnectionHandler(StreamSession session, int incomingSocketTimeout, boolean isPreview)
-    {
-        this.session = session;
-        this.isPreview = isPreview;
-        this.incoming = new IncomingMessageHandler(session, incomingSocketTimeout);
-        this.outgoing = new OutgoingMessageHandler(session);
-    }
-
-    /**
-     * Set up incoming message handler and initiate streaming.
-     *
-     * This method is called once on initiator.
-     *
-     * @throws IOException
-     */
-    @SuppressWarnings("resource")
-    public void initiate() throws IOException
-    {
-        logger.debug("[Stream #{}] Sending stream init for incoming stream", session.planId());
-        Socket incomingSocket = session.createConnection();
-        incoming.start(incomingSocket, StreamMessage.CURRENT_VERSION, true);
-
-        logger.debug("[Stream #{}] Sending stream init for outgoing stream", session.planId());
-        Socket outgoingSocket = session.createConnection();
-        outgoing.start(outgoingSocket, StreamMessage.CURRENT_VERSION, true);
-    }
-
-    /**
-     * Set up outgoing message handler on receiving side.
-     *
-     * @param connection Incoming connection to use for {@link OutgoingMessageHandler}.
-     * @param version Streaming message version
-     * @throws IOException
-     */
-    public void initiateOnReceivingSide(IncomingStreamingConnection connection, boolean isForOutgoing, int version) throws IOException
-    {
-        if (isForOutgoing)
-            outgoing.start(connection, version);
-        else
-            incoming.start(connection, version);
-    }
-
-    public ListenableFuture<?> close()
-    {
-        logger.debug("[Stream #{}] Closing stream connection handler on {}", session.planId(), session.peer);
-
-        ListenableFuture<?> inClosed = closeIncoming();
-        ListenableFuture<?> outClosed = closeOutgoing();
-
-        return Futures.allAsList(inClosed, outClosed);
-    }
-
-    public ListenableFuture<?> closeOutgoing()
-    {
-        return outgoing == null ? Futures.immediateFuture(null) : outgoing.close();
-    }
-
-    public ListenableFuture<?> closeIncoming()
-    {
-        return incoming == null ? Futures.immediateFuture(null) : incoming.close();
-    }
-
-    /**
-     * Enqueue messages to be sent.
-     *
-     * @param messages messages to send
-     */
-    public void sendMessages(Collection<? extends StreamMessage> messages)
-    {
-        for (StreamMessage message : messages)
-            sendMessage(message);
-    }
-
-    public void sendMessage(StreamMessage message)
-    {
-        if (outgoing.isClosed())
-            throw new RuntimeException("Outgoing stream handler has been closed");
-
-        if (message.type == StreamMessage.Type.FILE && isPreview)
-            throw new RuntimeException("Cannot send file messages for preview streaming sessions");
-
-        outgoing.enqueue(message);
-    }
-
-    /**
-     * @return true if outgoing connection is opened and ready to send messages
-     */
-    public boolean isOutgoingConnected()
-    {
-        return outgoing != null && !outgoing.isClosed();
-    }
-
-    abstract static class MessageHandler implements Runnable
-    {
-        protected final StreamSession session;
-
-        protected int protocolVersion;
-        private final boolean isOutgoingHandler;
-        protected Socket socket;
-
-        private final AtomicReference<SettableFuture<?>> closeFuture = new AtomicReference<>();
-        private IncomingStreamingConnection incomingConnection;
-
-        protected MessageHandler(StreamSession session, boolean isOutgoingHandler)
-        {
-            this.session = session;
-            this.isOutgoingHandler = isOutgoingHandler;
-        }
-
-        protected abstract String name();
-
-        @SuppressWarnings("resource")
-        protected static DataOutputStreamPlus getWriteChannel(Socket socket) throws IOException
-        {
-            WritableByteChannel out = socket.getChannel();
-            // socket channel is null when encrypted(SSL)
-            if (out == null)
-                return new WrappedDataOutputStreamPlus(new BufferedOutputStream(socket.getOutputStream()));
-            return new BufferedDataOutputStreamPlus(out);
-        }
-
-        protected static ReadableByteChannel getReadChannel(Socket socket) throws IOException
-        {
-            //we do this instead of socket.getChannel() so socketSoTimeout is respected
-            return Channels.newChannel(socket.getInputStream());
-        }
-
-        @SuppressWarnings("resource")
-        private void sendInitMessage() throws IOException
-        {
-            StreamInitMessage message = new StreamInitMessage(FBUtilities.getBroadcastAddress(),
-                                                              session.sessionIndex(),
-                                                              session.planId(),
-                                                              session.streamOperation(),
-                                                              !isOutgoingHandler,
-                                                              session.keepSSTableLevel(),
-                                                              session.getPendingRepair(),
-                                                              session.getPreviewKind());
-            ByteBuffer messageBuf = message.createMessage(false, protocolVersion);
-            DataOutputStreamPlus out = getWriteChannel(socket);
-            out.write(messageBuf);
-            out.flush();
-        }
-
-        public void start(IncomingStreamingConnection connection, int protocolVersion) throws IOException
-        {
-            this.incomingConnection = connection;
-            start(connection.socket, protocolVersion, false);
-        }
-
-        public void start(Socket socket, int protocolVersion, boolean initiator) throws IOException
-        {
-            this.socket = socket;
-            this.protocolVersion = protocolVersion;
-            if (initiator)
-                sendInitMessage();
-
-            new FastThreadLocalThread(this, name() + "-" + socket.getRemoteSocketAddress()).start();
-        }
-
-        public ListenableFuture<?> close()
-        {
-            // Assume it wasn't closed. Not a huge deal if we create a future on a race
-            SettableFuture<?> future = SettableFuture.create();
-            return closeFuture.compareAndSet(null, future)
-                 ? future
-                 : closeFuture.get();
-        }
-
-        public boolean isClosed()
-        {
-            return closeFuture.get() != null;
-        }
-
-        protected void signalCloseDone()
-        {
-            if (!isClosed())
-                close();
-
-            closeFuture.get().set(null);
-
-            // We can now close the socket
-            if (incomingConnection != null)
-            {
-                //this will close the underlying socket and remove it
-                //from active MessagingService connections (CASSANDRA-11854)
-                incomingConnection.close();
-            }
-            else
-            {
-                //this is an outgoing connection not registered in the MessagingService
-                //so we can close the socket directly
-                try
-                {
-                    socket.close();
-                }
-                catch (IOException e)
-                {
-                    // Erroring out while closing shouldn't happen but is not really a big deal, so just log
-                    // it at DEBUG and ignore otherwise.
-                    logger.debug("Unexpected error while closing streaming connection", e);
-                }
-            }
-        }
-    }
-
-    /**
-     * Incoming streaming message handler
-     */
-    static class IncomingMessageHandler extends MessageHandler
-    {
-        private final int socketTimeout;
-
-        IncomingMessageHandler(StreamSession session, int socketTimeout)
-        {
-            super(session, false);
-            this.socketTimeout = socketTimeout;
-        }
-
-        @Override
-        public void start(Socket socket, int version, boolean initiator) throws IOException
-        {
-            try
-            {
-                socket.setSoTimeout(socketTimeout);
-            }
-            catch (SocketException e)
-            {
-                logger.warn("Could not set incoming socket timeout to {}", socketTimeout, e);
-            }
-            super.start(socket, version, initiator);
-        }
-
-        protected String name()
-        {
-            return "STREAM-IN";
-        }
-
-        @SuppressWarnings("resource")
-        public void run()
-        {
-            try
-            {
-                ReadableByteChannel in = getReadChannel(socket);
-                while (!isClosed())
-                {
-                    // receive message
-                    StreamMessage message = StreamMessage.deserialize(in, protocolVersion, session);
-                    logger.debug("[Stream #{}] Received {}", session.planId(), message);
-                    // Might be null if there is an error during streaming (see FileMessage.deserialize). It's ok
-                    // to ignore here since we'll have asked for a retry.
-                    if (message != null)
-                    {
-                        session.messageReceived(message);
-                    }
-                }
-            }
-            catch (Throwable t)
-            {
-                JVMStabilityInspector.inspectThrowable(t);
-                session.onError(t);
-            }
-            finally
-            {
-                signalCloseDone();
-            }
-        }
-    }
-
-    /**
-     * Outgoing file transfer thread
-     */
-    static class OutgoingMessageHandler extends MessageHandler
-    {
-        /*
-         * All out going messages are queued up into messageQueue.
-         * The size will grow when received streaming request.
-         *
-         * Queue is also PriorityQueue so that prior messages can go out fast.
-         */
-        private final PriorityBlockingQueue<StreamMessage> messageQueue = new PriorityBlockingQueue<>(64, new Comparator<StreamMessage>()
-        {
-            public int compare(StreamMessage o1, StreamMessage o2)
-            {
-                return o2.getPriority() - o1.getPriority();
-            }
-        });
-
-        OutgoingMessageHandler(StreamSession session)
-        {
-            super(session, true);
-        }
-
-        protected String name()
-        {
-            return "STREAM-OUT";
-        }
-
-        public void enqueue(StreamMessage message)
-        {
-            messageQueue.put(message);
-        }
-
-        @SuppressWarnings("resource")
-        public void run()
-        {
-            try
-            {
-                DataOutputStreamPlus out = getWriteChannel(socket);
-
-                StreamMessage next;
-                while (!isClosed())
-                {
-                    if ((next = messageQueue.poll(1, TimeUnit.SECONDS)) != null)
-                    {
-                        logger.debug("[Stream #{}] Sending {}", session.planId(), next);
-                        sendMessage(out, next);
-                        if (next.type == StreamMessage.Type.SESSION_FAILED)
-                            close();
-                    }
-                }
-
-                // Sends the last messages on the queue
-                while ((next = messageQueue.poll()) != null)
-                    sendMessage(out, next);
-            }
-            catch (InterruptedException e)
-            {
-                throw new AssertionError(e);
-            }
-            catch (Throwable e)
-            {
-                session.onError(e);
-            }
-            finally
-            {
-                signalCloseDone();
-            }
-        }
-
-        private void sendMessage(DataOutputStreamPlus out, StreamMessage message)
-        {
-            try
-            {
-                StreamMessage.serialize(message, out, protocolVersion, session);
-                out.flush();
-                message.sent();
-            }
-            catch (SocketException e)
-            {
-                session.onError(e);
-                close();
-            }
-            catch (IOException e)
-            {
-                session.onError(e);
-            }
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java b/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
index d88d63c..d9ed8be 100644
--- a/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
+++ b/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
@@ -15,83 +15,93 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.apache.cassandra.streaming;
 
 import java.io.IOException;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.nio.channels.SocketChannel;
+import java.util.concurrent.TimeUnit;
+
+import javax.annotation.Nullable;
 
+import com.google.common.util.concurrent.Uninterruptibles;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import org.apache.cassandra.config.Config;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.WriteBufferWaterMark;
 import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.net.MessagingService;
-import org.apache.cassandra.security.SSLFactory;
-import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions;
+import org.apache.cassandra.net.async.NettyFactory;
+import org.apache.cassandra.net.async.OutboundConnectionIdentifier;
+import org.apache.cassandra.net.async.OutboundConnectionParams;
 
 public class DefaultConnectionFactory implements StreamConnectionFactory
 {
     private static final Logger logger = LoggerFactory.getLogger(DefaultConnectionFactory.class);
 
+    private static final int DEFAULT_CHANNEL_BUFFER_SIZE = 1 << 22;
+
+    private static final long MAX_WAIT_TIME_NANOS = TimeUnit.SECONDS.toNanos(30);
     private static final int MAX_CONNECT_ATTEMPTS = 3;
 
-    /**
-     * Connect to peer and start exchanging message.
-     * When connect attempt fails, this retries for maximum of MAX_CONNECT_ATTEMPTS times.
-     *
-     * @param peer the peer to connect to.
-     * @return the created socket.
-     *
-     * @throws IOException when connection failed.
-     */
-    public Socket createConnection(InetAddress peer) throws IOException
+    @Override
+    public Channel createConnection(OutboundConnectionIdentifier connectionId, int protocolVersion) throws IOException
     {
-        int attempts = 0;
+        ServerEncryptionOptions encryptionOptions = DatabaseDescriptor.getServerEncryptionOptions();
+
+        if (encryptionOptions.internode_encryption == ServerEncryptionOptions.InternodeEncryption.none)
+            encryptionOptions = null;
+
+        return createConnection(connectionId, protocolVersion, encryptionOptions);
+    }
+
+    protected Channel createConnection(OutboundConnectionIdentifier connectionId, int protocolVersion, @Nullable ServerEncryptionOptions encryptionOptions) throws IOException
+    {
+        // this is the amount of data to allow in memory before netty sets the channel writablility flag to false
+        int channelBufferSize = DEFAULT_CHANNEL_BUFFER_SIZE;
+        WriteBufferWaterMark waterMark = new WriteBufferWaterMark(channelBufferSize >> 2, channelBufferSize);
+
+        int sendBufferSize = DatabaseDescriptor.getInternodeSendBufferSize() > 0
+                             ? DatabaseDescriptor.getInternodeSendBufferSize()
+                             : OutboundConnectionParams.DEFAULT_SEND_BUFFER_SIZE;
+
+        OutboundConnectionParams params = OutboundConnectionParams.builder()
+                                                                  .connectionId(connectionId)
+                                                                  .encryptionOptions(encryptionOptions)
+                                                                  .mode(NettyFactory.Mode.STREAMING)
+                                                                  .protocolVersion(protocolVersion)
+                                                                  .sendBufferSize(sendBufferSize)
+                                                                  .waterMark(waterMark)
+                                                                  .build();
+
+        Bootstrap bootstrap = NettyFactory.instance.createOutboundBootstrap(params);
+
+        int connectionAttemptCount = 0;
+        long now = System.nanoTime();
+        final long end = now + MAX_WAIT_TIME_NANOS;
+        final Channel channel;
         while (true)
         {
-            try
-            {
-                Socket socket = newSocket(peer);
-                socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
-                socket.setKeepAlive(true);
-                return socket;
-            }
-            catch (IOException e)
+            ChannelFuture channelFuture = bootstrap.connect();
+            channelFuture.awaitUninterruptibly(end - now, TimeUnit.MILLISECONDS);
+            if (channelFuture.isSuccess())
             {
-                if (++attempts >= MAX_CONNECT_ATTEMPTS)
-                    throw e;
-
-                long waitms = DatabaseDescriptor.getRpcTimeout() * (long)Math.pow(2, attempts);
-                logger.warn("Failed attempt {} to connect to {}. Retrying in {} ms. ({})", attempts, peer, waitms, e.getMessage());
-                try
-                {
-                    Thread.sleep(waitms);
-                }
-                catch (InterruptedException wtf)
-                {
-                    throw new IOException("interrupted", wtf);
-                }
+                channel = channelFuture.channel();
+                break;
             }
-        }
-    }
 
-    // TODO this is deliberately copied from (the now former) OutboundTcpConnectionPool, for CASSANDRA-8457.
-    // to be replaced in CASSANDRA-12229 (make streaming use 8457)
-    public static Socket newSocket(InetAddress endpoint) throws IOException
-    {
-        // zero means 'bind on any available port.'
-        if (MessagingService.isEncryptedConnection(endpoint))
-        {
-            return SSLFactory.getSocket(DatabaseDescriptor.getServerEncryptionOptions(), endpoint, DatabaseDescriptor.getSSLStoragePort());
-        }
-        else
-        {
-            SocketChannel channel = SocketChannel.open();
-            channel.connect(new InetSocketAddress(endpoint, DatabaseDescriptor.getStoragePort()));
-            return channel.socket();
+            connectionAttemptCount++;
+            now = System.nanoTime();
+            if (connectionAttemptCount == MAX_CONNECT_ATTEMPTS || end - now <= 0)
+                throw new IOException("failed to connect to " + connectionId + " for streaming data", channelFuture.cause());
+
+            long waitms = DatabaseDescriptor.getRpcTimeout() * (long)Math.pow(2, connectionAttemptCount);
+            logger.warn("Failed attempt {} to connect to {}. Retrying in {} ms.", connectionAttemptCount, connectionId, waitms);
+            Uninterruptibles.sleepUninterruptibly(waitms, TimeUnit.MILLISECONDS);
         }
+
+        return channel;
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java b/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java
index dd99611..4cfe41e 100644
--- a/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java
+++ b/src/java/org/apache/cassandra/streaming/StreamConnectionFactory.java
@@ -15,16 +15,15 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.apache.cassandra.streaming;
 
 import java.io.IOException;
-import java.net.InetAddress;
-import java.net.Socket;
 
-/**
- * Interface that creates connection used by streaming.
- */
+import io.netty.channel.Channel;
+import org.apache.cassandra.net.async.OutboundConnectionIdentifier;
+
 public interface StreamConnectionFactory
 {
-    Socket createConnection(InetAddress peer) throws IOException;
+    Channel createConnection(OutboundConnectionIdentifier connectionId, int protocolVersion) throws IOException;
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/streaming/StreamCoordinator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamCoordinator.java b/src/java/org/apache/cassandra/streaming/StreamCoordinator.java
index 9059f45..bb8c702 100644
--- a/src/java/org/apache/cassandra/streaming/StreamCoordinator.java
+++ b/src/java/org/apache/cassandra/streaming/StreamCoordinator.java
@@ -37,8 +37,10 @@ public class StreamCoordinator
 {
     private static final Logger logger = LoggerFactory.getLogger(StreamCoordinator.class);
 
-    // Executor strictly for establishing the initial connections. Once we're connected to the other end the rest of the
-    // streaming is handled directly by the ConnectionHandler's incoming and outgoing threads.
+    /**
+     * Executor strictly for establishing the initial connections. Once we're connected to the other end the rest of the
+     * streaming is handled directly by the {@link StreamingMessageSender}'s incoming and outgoing threads.
+     */
     private static final DebuggableThreadPoolExecutor streamExecutor = DebuggableThreadPoolExecutor.createWithFixedPoolSize("StreamConnectionEstablisher",
                                                                                                                             FBUtilities.getAvailableProcessors());
     private final boolean connectSequentially;
@@ -55,8 +57,8 @@ public class StreamCoordinator
                              boolean connectSequentially, UUID pendingRepair, PreviewKind previewKind)
     {
         this.connectionsPerHost = connectionsPerHost;
-        this.factory = factory;
         this.keepSSTableLevel = keepSSTableLevel;
+        this.factory = factory;
         this.connectSequentially = connectSequentially;
         this.pendingRepair = pendingRepair;
         this.previewKind = previewKind;
@@ -163,6 +165,11 @@ public class StreamCoordinator
         return getOrCreateHostData(peer).getOrCreateSessionById(peer, id, connecting);
     }
 
+    public StreamSession getSessionById(InetAddress peer, int id)
+    {
+        return getHostData(peer).getSessionById(id);
+    }
+
     public synchronized void updateProgress(ProgressInfo info)
     {
         getHostData(info.peer).updateProgress(info);
@@ -274,8 +281,8 @@ public class StreamCoordinator
 
     private class HostStreamingData
     {
-        private Map<Integer, StreamSession> streamSessions = new HashMap<>();
-        private Map<Integer, SessionInfo> sessionInfos = new HashMap<>();
+        private final Map<Integer, StreamSession> streamSessions = new HashMap<>();
+        private final Map<Integer, SessionInfo> sessionInfos = new HashMap<>();
 
         private int lastReturned = -1;
 
@@ -333,6 +340,11 @@ public class StreamCoordinator
             return session;
         }
 
+        public StreamSession getSessionById(int id)
+        {
+            return streamSessions.get(id);
+        }
+
         public void updateProgress(ProgressInfo info)
         {
             sessionInfos.get(info.sessionIndex).updateProgress(info);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/streaming/StreamManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamManager.java b/src/java/org/apache/cassandra/streaming/StreamManager.java
index 52652c0..a44f02e 100644
--- a/src/java/org/apache/cassandra/streaming/StreamManager.java
+++ b/src/java/org/apache/cassandra/streaming/StreamManager.java
@@ -21,7 +21,6 @@ import java.net.InetAddress;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
-
 import javax.management.ListenerNotFoundException;
 import javax.management.MBeanNotificationInfo;
 import javax.management.NotificationFilter;
@@ -136,7 +135,7 @@ public class StreamManager implements StreamManagerMBean
         initiatedStreams.put(result.planId, result);
     }
 
-    public void registerReceiving(final StreamResultFuture result)
+    public StreamResultFuture registerReceiving(final StreamResultFuture result)
     {
         result.addEventListener(notifier);
         // Make sure we remove the stream on completion (whether successful or not)
@@ -148,7 +147,8 @@ public class StreamManager implements StreamManagerMBean
             }
         }, MoreExecutors.directExecutor());
 
-        receivingStreams.put(result.planId, result);
+        StreamResultFuture previous = receivingStreams.putIfAbsent(result.planId, result);
+        return previous ==  null ? result : previous;
     }
 
     public StreamResultFuture getReceivingStream(UUID planId)
@@ -175,4 +175,22 @@ public class StreamManager implements StreamManagerMBean
     {
         return notifier.getNotificationInfo();
     }
+
+    public StreamSession findSession(InetAddress peer, UUID planId, int sessionIndex)
+    {
+        StreamSession session = findSession(initiatedStreams, peer, planId, sessionIndex);
+        if (session !=  null)
+            return session;
+
+        return findSession(receivingStreams, peer, planId, sessionIndex);
+    }
+
+    private StreamSession findSession(Map<UUID, StreamResultFuture> streams, InetAddress peer, UUID planId, int sessionIndex)
+    {
+        StreamResultFuture streamResultFuture = streams.get(planId);
+        if (streamResultFuture == null)
+            return null;
+
+        return streamResultFuture.getSession(peer, sessionIndex);
+    }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fc92db2b/src/java/org/apache/cassandra/streaming/StreamPlan.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamPlan.java b/src/java/org/apache/cassandra/streaming/StreamPlan.java
index 05a8d30..213f74b 100644
--- a/src/java/org/apache/cassandra/streaming/StreamPlan.java
+++ b/src/java/org/apache/cassandra/streaming/StreamPlan.java
@@ -33,7 +33,7 @@ import static org.apache.cassandra.service.ActiveRepairService.NO_PENDING_REPAIR
  */
 public class StreamPlan
 {
-    public static final String[] EMPTY_COLUMN_FAMILIES = new String[0];
+    private static final String[] EMPTY_COLUMN_FAMILIES = new String[0];
     private final UUID planId = UUIDGen.getTimeUUID();
     private final StreamOperation streamOperation;
     private final List<StreamEventHandler> handlers = new ArrayList<>();


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org


Mime
View raw message