From 709b9fc319f669ee07338a842f36a9d77e8c46a1 Mon Sep 17 00:00:00 2001
From: Marcus Eriksson <marcuse@apache.org>
Date: Mon, 2 Jun 2014 08:53:32 +0200
Subject: [PATCH] Make StreamSession#closeSession() idempotent

Patch by JoshuaMcKenzie; reviewed by marcuse for CASSANDRA-7262
---
 CHANGES.txt                                   |  1 +
 .../cassandra/streaming/StreamSession.java    | 34 +++++++++++--------
 2 files changed, 21 insertions(+), 14 deletions(-)

diff --git a/CHANGES.txt b/CHANGES.txt
index d1d10301d8..bc95a8dd2e 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -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)
diff --git a/src/java/org/apache/cassandra/streaming/StreamSession.java b/src/java/org/apache/cassandra/streaming/StreamSession.java
index 30e3fa2b38..79ad4870b6 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -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);
     }
 
     /**