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

Merge branch 'cassandra-2.0' into cassandra-2.1 

Conflicts:
	CHANGES.txt
This commit is contained in:
Marcus Eriksson 2014-06-02 08:59:50 +02:00
parent 1c33ea3677 709b9fc319
commit e398c6bb89
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

@ -2,6 +2,7 @@
* Upgrade to Pig 0.12.1 (CASSANDRA-6556)
Merged from 2.0:
* Fix NPE in StreamTransferTask.createMessageForRetry() (CASSANDRA-7323)
* Make StreamSession#closeSession() idempotent (CASSANDRA-7262)
Merged from 1.2:
* 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;
@ -128,6 +129,8 @@ public class StreamSession implements IEndpointStateChangeSubscriber, IFailureDe
private int retries;
private AtomicBoolean isAborted = new AtomicBoolean(false);
public static enum State
{
INITIALIZED,
@ -339,23 +342,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);
}
/**