diff --git a/CHANGES.txt b/CHANGES.txt index 6fd55e6d73..0fe2285803 100644 --- a/CHANGES.txt +++ b/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 diff --git a/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java b/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java index e875ee338c..c8d3723e6a 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java +++ b/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) { diff --git a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java index 221b3c1eec..33f2ac1342 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java +++ b/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; + } + } } /** diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java index c52184bedd..2d34209b8e 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java +++ b/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 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. */ diff --git a/src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java b/src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java index b95bf32bcf..f4281b2d5c 100644 --- a/src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java +++ b/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(); + } } diff --git a/src/java/org/apache/cassandra/io/util/SequentialWriter.java b/src/java/org/apache/cassandra/io/util/SequentialWriter.java index aea97aa9bb..462eb0a352 100644 --- a/src/java/org/apache/cassandra/io/util/SequentialWriter.java +++ b/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