Merge branch 'cassandra-2.1' into cassandra-2.2

CHANGES.txt

@@ -27,6 +27,7 @@
  * Fix paging on DISTINCT queries repeats result when first row in partition changes
    (CASSANDRA-10010)
 Merged from 2.1:
+ * Fix streaming_socket_timeout_in_ms not enforced (CASSANDRA-11286)
  * Avoid dropping message too quickly due to missing unit conversion (CASSANDRA-11302)
  * COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053)
  * InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)

conf/cassandra.yaml

@@ -671,10 +671,9 @@ request_timeout_in_ms: 10000
 # and the times are synchronized between the nodes.
 cross_node_timeout: false
 
-# Enable socket timeout for streaming operation.
-# When a timeout occurs during streaming, streaming is retried from the start
-# of the current file. This _can_ involve re-streaming an important amount of
-# data, so you should avoid setting the value too low.
+# Set socket timeout for streaming operation.
+# The stream session is failed if no data is received by any of the
+# participants within that period.
 # Default value is 3600000, which means streams timeout after an hour.
 # streaming_socket_timeout_in_ms: 3600000
 

src/java/org/apache/cassandra/net/IncomingStreamingConnection.java

@@ -27,6 +27,7 @@ import java.util.Set;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+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;
@@ -62,6 +63,10 @@ public class IncomingStreamingConnection extends Thread implements Closeable
             DataInput input = new DataInputStream(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
@@ -74,7 +79,7 @@ public class IncomingStreamingConnection extends Thread implements Closeable
             close();
         }
     }
-    
+
     @Override
     public void close()
     {

src/java/org/apache/cassandra/streaming/ConnectionHandler.java

@@ -109,12 +109,22 @@ public class ConnectionHandler
     {
         logger.debug("[Stream #{}] Closing stream connection handler on {}", session.planId(), session.peer);
 
-        ListenableFuture<?> inClosed = incoming == null ? Futures.immediateFuture(null) : incoming.close();
-        ListenableFuture<?> outClosed = outgoing == null ? Futures.immediateFuture(null) : outgoing.close();
+        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.
      *
@@ -170,11 +180,8 @@ public class ConnectionHandler
 
         protected static ReadableByteChannel getReadChannel(Socket socket) throws IOException
         {
-            ReadableByteChannel in = socket.getChannel();
-            // socket channel is null when encrypted(SSL)
-            return in == null
-                 ? Channels.newChannel(socket.getInputStream())
-                 : in;
+            //we do this instead of socket.getChannel() so socketSoTimeout is respected
+            return Channels.newChannel(socket.getInputStream());
         }
 
         @SuppressWarnings("resource")

src/java/org/apache/cassandra/streaming/StreamSession.java

@@ -628,6 +628,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
         else
         {
             state(State.WAIT_COMPLETE);
+            handler.closeIncoming();
         }
     }
 
@@ -715,6 +716,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
                 handler.sendMessage(new CompleteMessage());
                 completeSent = true;
                 state(State.WAIT_COMPLETE);
+                handler.closeOutgoing();
             }
         }
         return completed;