Merge branch 'cassandra-3.0' into cassandra-3.7

CHANGES.txt

@@ -13,6 +13,7 @@ Merged from 2.2:
  * Prohibit Reversed Counter type as part of the PK (CASSANDRA-9395)
  * cqlsh: correctly handle non-ascii chars in error messages (CASSANDRA-11626)
 Merged from 2.1:
+ * Set default streaming_socket_timeout_in_ms to 24 hours (CASSANDRA-11840)
  * Do not consider local node a valid source during replace (CASSANDRA-11848)
  * Add message dropped tasks to nodetool netstats (CASSANDRA-11855)
  * Avoid holding SSTableReaders for duration of incremental repair (CASSANDRA-11739)

conf/cassandra.yaml

@@ -774,10 +774,12 @@ request_timeout_in_ms: 10000
 cross_node_timeout: false
 
 # Set socket timeout for streaming operation.
-# The stream session is failed if no data is received by any of the
+# The stream session is failed if no data/ack is received by any of the participants
-# participants within that period.
+# within that period, which means this should also be sufficient to stream a large
-# Default value is 3600000, which means streams timeout after an hour.
+# sstable or rebuild table indexes.
-# streaming_socket_timeout_in_ms: 3600000
+# Default value is 86400000ms, which means stale streams timeout after 24 hours.
+# A value of zero means stream sockets should never time out.
+# streaming_socket_timeout_in_ms: 86400000
 
 # phi value that must be reached for a host to be marked down.
 # most users should never need to adjust this.

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

@@ -99,7 +99,7 @@ public class Config
 
     public volatile Long truncate_request_timeout_in_ms = 60000L;
 
-    public Integer streaming_socket_timeout_in_ms = 3600000;
+    public Integer streaming_socket_timeout_in_ms = 86400000; //24 hours
 
     public boolean cross_node_timeout = false;
 

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

@@ -20,6 +20,7 @@ package org.apache.cassandra.streaming;
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.Socket;
+import java.net.SocketTimeoutException;
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -516,10 +517,20 @@ public class StreamSession implements IEndpointStateChangeSubscriber
      */
     public void onError(Throwable e)
     {
-        logger.error("[Stream #{}] Streaming error occurred on session with peer {}{}", planId(),
+        if (e instanceof SocketTimeoutException)
-                                                                                        peer.getHostAddress(),
+        {
-                                                                                        peer.equals(connecting) ? "" : " through " + connecting.getHostAddress(),
+            logger.error("[Stream #{}] Streaming socket timed out. This means the session peer stopped responding or " +
-                                                                                        e);
+                         "is still processing received data. If there is no sign of failure in the other end or a very " +
+                         "dense table is being transferred you may want to increase streaming_socket_timeout_in_ms " +
+                         "property. Current value is {}ms.", planId(), DatabaseDescriptor.getStreamingSocketTimeout(), e);
+        }
+        else
+        {
+            logger.error("[Stream #{}] Streaming error occurred on session with peer {}{}", planId(),
+                                                                                            peer.getHostAddress(),
+                                                                                            peer.equals(connecting) ? "" : " through " + connecting.getHostAddress(),
+                                                                                            e);
+        }
         // send session failure message
         if (handler.isOutgoingConnected())
             handler.sendMessage(new SessionFailedMessage());