natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Backport CASSANDRA-3569 

patch by Omid Aladini;reviewed by yukim for CASSANDRA-9455
This commit is contained in:
Omid Aladini 2015-05-26 13:16:50 -05:00 committed by Yuki Morishita
parent 04287d440a
commit 3e4ed96663
4 changed files with 5 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -5,6 +5,7 @@
* Push correct protocol notification for DROP INDEX (CASSANDRA-9310) * Push correct protocol notification for DROP INDEX (CASSANDRA-9310)
* token-generator - generated tokens too long (CASSANDRA-9300) * token-generator - generated tokens too long (CASSANDRA-9300)
* Add option not to validate atoms during scrub (CASSANDRA-9406) * Add option not to validate atoms during scrub (CASSANDRA-9406)
* Backport CASSANDRA-3569 (CASSANDRA-9455)
2.0.15: 2.0.15:

1
debian/cassandra-sysctl.conf vendored
View File

@ -1 +1,2 @@
vm.max_map_count = 1048575 vm.max_map_count = 1048575
net.ipv4.tcp_keepalive_time=300

1
src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java
View File

@ -51,6 +51,7 @@ public class DefaultConnectionFactory implements StreamConnectionFactory
{ {
Socket socket = OutboundTcpConnectionPool.newSocket(peer); Socket socket = OutboundTcpConnectionPool.newSocket(peer);
socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout()); socket.setSoTimeout(DatabaseDescriptor.getStreamingSocketTimeout());
socket.setKeepAlive(true);
return socket; return socket;
} }
catch (IOException e) catch (IOException e)

22
src/java/org/apache/cassandra/streaming/StreamSession.java
View File

@ -109,7 +109,7 @@ import org.apache.cassandra.utils.Pair;
* session is done is is closed (closeSession()). Otherwise, the node switch to the WAIT_COMPLETE state and * session is done is is closed (closeSession()). Otherwise, the node switch to the WAIT_COMPLETE state and
* send a CompleteMessage to the other side. * send a CompleteMessage to the other side.
*/ */
public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDetectionEventListener public class StreamSession implements IEndpointStateChangeSubscriber
{ {
private static final Logger logger = LoggerFactory.getLogger(StreamSession.class); private static final Logger logger = LoggerFactory.getLogger(StreamSession.class);
@ -194,10 +194,6 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
public void init(StreamResultFuture streamResult) public void init(StreamResultFuture streamResult)
{ {
this.streamResult = streamResult; this.streamResult = streamResult;
// register to gossiper/FD to fail on node failure
Gossiper.instance.register(this);
FailureDetector.instance.registerFailureDetectionEventListener(this);
} }
public void start() public void start()
@ -372,8 +368,6 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
// incoming thread (so we would deadlock). // incoming thread (so we would deadlock).
handler.close(); handler.close();
Gossiper.instance.unregister(this);
FailureDetector.instance.unregisterFailureDetectionEventListener(this);
streamResult.handleSessionComplete(this); streamResult.handleSessionComplete(this);
} }
} }
@ -627,23 +621,11 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
public void onRemove(InetAddress endpoint) public void onRemove(InetAddress endpoint)
{ {
convict(endpoint, Double.MAX_VALUE); closeSession(State.FAILED);
} }
public void onRestart(InetAddress endpoint, EndpointState epState) public void onRestart(InetAddress endpoint, EndpointState epState)
{ {
convict(endpoint, Double.MAX_VALUE);
}
public void convict(InetAddress endpoint, double phi)
{
if (!endpoint.equals(peer))
return;
// We want a higher confidence in the failure detection than usual because failing a streaming wrongly has a high cost (CASSANDRA-7063)
if (phi < 100 * DatabaseDescriptor.getPhiConvictThreshold())
return;
closeSession(State.FAILED); closeSession(State.FAILED);
} }