diff --git a/CHANGES.txt b/CHANGES.txt
index af4da65285..c7c6877b45 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -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)
 
diff --git a/src/java/org/apache/cassandra/streaming/StreamSession.java b/src/java/org/apache/cassandra/streaming/StreamSession.java
index 9f916d9f17..411f9695e7 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;
@@ -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);
     }
 
     /**