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

Make StreamSession#closeSession() idempotent 

Patch by JoshuaMcKenzie; reviewed by marcuse for CASSANDRA-7262
This commit is contained in:
Marcus Eriksson 2014-06-02 08:53:32 +02:00
parent e68ac31f3f
commit 709b9fc319
2 changed files with 21 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -10,6 +10,7 @@
* Don't try to compact already-compacting files in HHOM (CASSANDRA-7288)
* Add authentication support to shuffle (CASSANDRA-6484)
* Cqlsh counts non-empty lines for "Blank lines" warning (CASSANDRA-7325)
* Make StreamSession#closeSession() idempotent (CASSANDRA-7262)
Merged from 1.2:
* Fix availability validation for LOCAL_ONE CL (CASSANDRA-7319)
* Use LOCAL_ONE for non-superuser auth queries (CASSANDRA-7328)

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

@ -21,6 +21,7 @@ import java.io.IOException;
import java.net.InetAddress;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.collect.*;
import org.slf4j.Logger;
@ -132,6 +133,8 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
private int retries;
private AtomicBoolean isAborted = new AtomicBoolean(false);
public static enum State
{
INITIALIZED,
@ -329,23 +332,26 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
}
}
private void closeSession(State finalState)
private synchronized void closeSession(State finalState)
{
state(finalState);
if (finalState == State.FAILED)
if (isAborted.compareAndSet(false, true))
{
for (StreamTask task : Iterables.concat(receivers.values(), transfers.values()))
task.abort();
state(finalState);
if (finalState == State.FAILED)
{
for (StreamTask task : Iterables.concat(receivers.values(), transfers.values()))
task.abort();
}
// Note that we shouldn't block on this close because this method is called on the handler
// incoming thread (so we would deadlock).
handler.close();
Gossiper.instance.unregister(this);
FailureDetector.instance.unregisterFailureDetectionEventListener(this);
streamResult.handleSessionComplete(this);
}
// Note that we shouldn't block on this close because this method is called on the handler
// incoming thread (so we would deadlock).
handler.close();
Gossiper.instance.unregister(this);
FailureDetector.instance.unregisterFailureDetectionEventListener(this);
streamResult.handleSessionComplete(this);
}
/**