Fix streaming to catch exception so retry not fail

patch by yukim; reviewed by Paulo Motta for CASSANDRA-10557

CHANGES.txt

@@ -1,4 +1,5 @@
 2.1.12
+ * Fix streaming to catch exception so retry not fail (CASSANDRA-10557)
  * Add validation method to PerRowSecondaryIndex (CASSANDRA-10092)
  * Support encrypted and plain traffic on the same port (CASSANDRA-10559)
  * Do STCS in DTCS windows (CASSANDRA-10276)

src/java/org/apache/cassandra/streaming/StreamReader.java

@@ -89,11 +89,12 @@ public class StreamReader
         }
         ColumnFamilyStore cfs = Keyspace.open(kscf.left).getColumnFamilyStore(kscf.right);
 
-        SSTableWriter writer = createWriter(cfs, totalSize, repairedAt);
         DataInputStream dis = new DataInputStream(new LZFInputStream(Channels.newInputStream(channel)));
         BytesReadTracker in = new BytesReadTracker(dis);
+        SSTableWriter writer = null;
         try
         {
+            writer = createWriter(cfs, totalSize, repairedAt);
             while (in.getBytesRead() < totalSize)
             {
                 writeRow(writer, in, cfs);
@@ -104,7 +105,18 @@ public class StreamReader
         }
         catch (Throwable e)
         {
-            writer.abort();
+            if (writer != null)
+            {
+                try
+                {
+                    writer.abort();
+                }
+                catch (Throwable e2)
+                {
+                    // add abort error to original and continue so we can drain unread stream
+                    e.addSuppressed(e2);
+                }
+            }
             drain(dis, in.getBytesRead());
             if (e instanceof IOException)
                 throw (IOException) e;

src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java

@@ -72,12 +72,12 @@ public class CompressedStreamReader extends StreamReader
         }
         ColumnFamilyStore cfs = Keyspace.open(kscf.left).getColumnFamilyStore(kscf.right);
 
-        SSTableWriter writer = createWriter(cfs, totalSize, repairedAt);
-
         CompressedInputStream cis = new CompressedInputStream(Channels.newInputStream(channel), compressionInfo, inputVersion.hasPostCompressionAdlerChecksums);
         BytesReadTracker in = new BytesReadTracker(new DataInputStream(cis));
+        SSTableWriter writer = null;
         try
         {
+            writer = createWriter(cfs, totalSize, repairedAt);
             for (Pair<Long, Long> section : sections)
             {
                 long length = section.right - section.left;
@@ -95,7 +95,18 @@ public class CompressedStreamReader extends StreamReader
         }
         catch (Throwable e)
         {
-            writer.abort();
+            if (writer != null)
+            {
+                try
+                {
+                    writer.abort();
+                }
+                catch (Throwable e2)
+                {
+                    // add abort error to original and continue so we can drain unread stream
+                    e.addSuppressed(e2);
+                }
+            }
             drain(cis, in.getBytesRead());
             if (e instanceof IOException)
                 throw (IOException) e;