Merge branch 'cassandra-2.1' into trunk

CHANGES.txt

@@ -89,6 +89,9 @@
  * Log failed host when preparing incremental repair (CASSANDRA-8228)
  * Force config client mode in CQLSSTableWriter (CASSANDRA-8281)
 Merged from 2.0:
+=======
+2.0.12:
+ * Ensure SSTableWriter cleans up properly after failure (CASSANDRA-8499)
  * Increase bf true positive count on key cache hit (CASSANDRA-8525)
  * Move MeteredFlusher to its own thread (CASSANDRA-8485)
  * Fix non-distinct results in DISTNCT queries on static columns when

src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java

@@ -261,6 +261,12 @@ public class CompressedSequentialWriter extends SequentialWriter
         }
     }
 
+    public void abort()
+    {
+        super.abort();
+        metadataWriter.abort();
+    }
+
     @Override
     public void writeFullChecksum(Descriptor descriptor)
     {

src/java/org/apache/cassandra/io/compress/CompressionMetadata.java

@@ -381,6 +381,15 @@ public class CompressionMetadata
                 FileUtils.closeQuietly(out);
             }
         }
+
+        public void abort()
+        {
+            if (offsets != null)
+            {
+                offsets.unreference();
+                offsets = null;
+            }
+        }
     }
 
     /**

src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java

@@ -289,16 +289,12 @@ public class BigTableWriter extends SSTableWriter
         assert descriptor.type.isTemporary;
         if (iwriter == null && dataFile == null)
             return;
+
         if (iwriter != null)
-        {
-            FileUtils.closeQuietly(iwriter.indexFile);
-            if (closeBf)
-            {
-                iwriter.bf.close();
-            }
-        }
+            iwriter.abort(closeBf);
+
         if (dataFile!= null)
-            FileUtils.closeQuietly(dataFile);
+            dataFile.abort();
 
         Set<Component> components = SSTable.componentsFor(descriptor);
         try
@@ -498,7 +494,7 @@ public class BigTableWriter extends SSTableWriter
     /**
      * Encapsulates writing the index and filter for an SSTable. The state of this object is not valid until it has been closed.
      */
-    class IndexWriter implements Closeable
+    class IndexWriter
     {
         private final SequentialWriter indexFile;
         public final SegmentedFile.Builder builder;
@@ -542,6 +538,13 @@ public class BigTableWriter extends SSTableWriter
             builder.addPotentialBoundary(indexPosition);
         }
 
+        public void abort(boolean closeBf)
+        {
+            indexFile.abort();
+            if (closeBf)
+                bf.close();
+        }
+
         /**
          * Closes the index and bloomfilter, making the public state of this writer valid for consumption.
          */

src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java

@@ -50,4 +50,10 @@ public class ChecksummedSequentialWriter extends SequentialWriter
         super.close();
         crcWriter.close();
     }
+
+    public void abort()
+    {
+        super.abort();
+        crcWriter.abort();
+    }
 }

src/java/org/apache/cassandra/io/util/SequentialWriter.java

@@ -22,6 +22,9 @@ import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
 import java.nio.channels.WritableByteChannel;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.io.FSReadError;
 import org.apache.cassandra.io.FSWriteError;
@@ -38,6 +41,8 @@ import org.apache.cassandra.utils.CLibrary;
  */
 public class SequentialWriter extends OutputStream implements WritableByteChannel
 {
+    private static final Logger logger = LoggerFactory.getLogger(SequentialWriter.class);
+
     // isDirty - true if this.buffer contains any un-synced bytes
     protected boolean isDirty = false, syncNeeded = false;
 
@@ -46,7 +51,7 @@ public class SequentialWriter extends OutputStream implements WritableByteChanne
 
     protected byte[] buffer;
     private final int fd;
-    private final int directoryFD;
+    private int directoryFD;
     // directory should be synced only after first file sync, in other words, only once per file
     private boolean directorySynced = false;
 
@@ -439,16 +444,34 @@ public class SequentialWriter extends OutputStream implements WritableByteChanne
 
         buffer = null;
 
-        try
+        cleanup(true);
+    }
+
+    public void abort()
+    {
+        cleanup(false);
+    }
+
+    private void cleanup(boolean throwExceptions)
+    {
+        if (directoryFD >= 0)
         {
-            out.close();
-        }
-        catch (IOException e)
-        {
-            throw new FSWriteError(e, getPath());
+            try { CLibrary.tryCloseFD(directoryFD); }
+            catch (Throwable t) { handle(t, throwExceptions); }
+            directoryFD = -1;
         }
 
-        CLibrary.tryCloseFD(directoryFD);
+        // close is idempotent
+        try { out.close(); }
+        catch (Throwable t) { handle(t, throwExceptions); }
+    }
+
+    private void handle(Throwable t, boolean throwExceptions)
+    {
+        if (!throwExceptions)
+            logger.warn("Suppressing exception thrown while aborting writer", t);
+        else
+            throw new FSWriteError(t, getPath());
     }
 
     // hack to make life easier for subclasses