Add optional socket timeout for streaming

patch by vijay2win; reviewed by slebresne for CASSANDRA-3838

CHANGES.txt

@@ -10,6 +10,7 @@
    readability (CASSANDRA-3726)
  * synchronize BiMap of bootstrapping tokens (CASSANDRA-3417)
  * show index options in CLI (CASSANDRA-3809)
+ * add optional socket timeout for streaming (CASSANDRA-3838)
 Merged from 0.8:
  * (Pig) fix CassandraStorage to use correct comparator in Super ColumnFamily
    case (CASSANDRA-3251)

NEWS.txt

@@ -8,6 +8,13 @@ upgrade, just in case you need to roll back to the previous version.
 (Cassandra version X + 1 will always be able to read data files created
 by version X, but the inverse is not necessarily the case.)
 
+1.0.8
+=====
+
+Other
+-----
+    - Allow configuring socket timeout for streaming
+
 1.0.7
 =====
 

conf/cassandra.yaml

@@ -315,6 +315,13 @@ compaction_preheat_key_cache: true
 # Time to wait for a reply from other nodes before failing the command 
 rpc_timeout_in_ms: 10000
 
+# 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.
+# Default value is 0, which never timeout streams.
+# streaming_socket_timeout_in_ms: 0
+
 # phi value that must be reached for a host to be marked down.
 # most users should never need to adjust this.
 # phi_convict_threshold: 8

src/java/org/apache/cassandra/config/Config.java

@@ -50,6 +50,8 @@ public class Config
     
     public Long rpc_timeout_in_ms = new Long(2000);
 
+    public Integer streaming_socket_timeout_in_ms = new Integer(0);
+
     public Integer phi_convict_threshold = 8;
     
     public Integer concurrent_reads = 8;

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -1008,4 +1008,9 @@ public class DatabaseDescriptor
     {
         return conf.commitlog_total_space_in_mb;
     }
+
+    public static int getStreamingSocketTimeout()
+    {
+        return conf.streaming_socket_timeout_in_ms;
+    }
 }

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

@@ -197,6 +197,7 @@ public class FileStreamTask extends WrappedRunnable
             try
             {
                 socket = MessagingService.instance().getConnectionPool(to).newSocket();
+                socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
                 output = socket.getOutputStream();
                 break;
             }

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

@@ -55,6 +55,7 @@ public class IncomingStreamReader
 
     public IncomingStreamReader(StreamHeader header, Socket socket) throws IOException
     {
+        socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
         this.socket = socket;
         InetSocketAddress remoteAddress = (InetSocketAddress)socket.getRemoteSocketAddress();
         session = StreamInSession.get(remoteAddress.getAddress(), header.sessionId);