Do not create sstable files before registering in txn

Refactoring prevents the situation where some sstable components, like
data or index, are created before the new sstable is registered with
lifecycle transaction, which leads to a problem such that there is
a short time when incomplete sstable components are present. At the same
time, no transaction file is created, which leads to the possibility
that the sstable can be recognized as completed by various
transaction-aware listers.

Patch by Jacek Lewandowski; reviewed by Branimir Lambov, Mike Adamson for CASSANDRA-18737

CHANGES.txt

@@ -1,4 +1,5 @@
 5.0-alpha2
+ * Do not create sstable files before registering in txn (CASSANDRA-18737)
  * Do not log stacktrace on mismatched cache and schema version and checksum error in AutoSavingCache (CASSANDRA-18862)
  * Remove wrong assertion in SSTableLoader (CASSANDRA-18840)
  * Fix accessing java.nio.Bits.TOTAL_CAPACITY in Java17 (CASSANDRA-18848)

src/java/org/apache/cassandra/db/compaction/CompactionManager.java

@@ -1624,7 +1624,7 @@ public class CompactionManager implements CompactionManagerMBean
                          .setMetadataCollector(new MetadataCollector(cfs.metadata().comparator).sstableLevel(sstable.getSSTableLevel()))
                          .setSerializationHeader(sstable.header)
                          .addDefaultComponents(cfs.indexManager.listIndexGroups())
-                         .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), txn, cfs.metadata.get())
+                         .setSecondaryIndexGroups(cfs.indexManager.listIndexGroups())
                          .build(txn, cfs);
     }
 
@@ -1664,7 +1664,7 @@ public class CompactionManager implements CompactionManagerMBean
                          .setMetadataCollector(new MetadataCollector(sstables, cfs.metadata().comparator).sstableLevel(minLevel))
                          .setSerializationHeader(SerializationHeader.make(cfs.metadata(), sstables))
                          .addDefaultComponents(cfs.indexManager.listIndexGroups())
-                         .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), txn, cfs.metadata.get())
+                         .setSecondaryIndexGroups(cfs.indexManager.listIndexGroups())
                          .build(txn, cfs);
     }
 

src/java/org/apache/cassandra/db/compaction/Upgrader.java

@@ -84,7 +84,7 @@ public class Upgrader
                          .setMetadataCollector(sstableMetadataCollector)
                          .setSerializationHeader(SerializationHeader.make(cfs.metadata(), Sets.newHashSet(sstable)))
                          .addDefaultComponents(cfs.indexManager.listIndexGroups())
-                         .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), transaction, cfs.metadata.get())
+                         .setSecondaryIndexGroups(cfs.indexManager.listIndexGroups())
                          .build(transaction, cfs);
     }
 

src/java/org/apache/cassandra/db/compaction/unified/ShardedMultiWriter.java

@@ -117,7 +117,7 @@ public class ShardedMultiWriter implements SSTableMultiWriter
                          .setMetadataCollector(metadataCollector)
                          .setSerializationHeader(header)
                          .addDefaultComponents(indexGroups)
-                         .addFlushObserversForSecondaryIndexes(indexGroups, lifecycleNewTracker, cfs.metadata.get())
+                         .setSecondaryIndexGroups(indexGroups)
                          .build(lifecycleNewTracker, cfs);
     }
 

src/java/org/apache/cassandra/db/compaction/writers/CompactionAwareWriter.java

@@ -319,7 +319,7 @@ public abstract class CompactionAwareWriter extends Transactional.AbstractTransa
                          .setTransientSSTable(isTransient)
                          .setRepairedAt(minRepairedAt)
                          .setPendingRepair(pendingRepair)
-                         .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), txn, cfs.metadata.get())
+                         .setSecondaryIndexGroups(cfs.indexManager.listIndexGroups())
                          .addDefaultComponents(cfs.indexManager.listIndexGroups());
     }
 }

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

@@ -202,7 +202,7 @@ public class CompressedSequentialWriter extends SequentialWriter
         // next chunk should be written right after current + length of the checksum (int)
         chunkOffset += compressedLength + 4;
         if (runPostFlush != null)
-            runPostFlush.run();
+            runPostFlush.accept(getLastFlushOffset());
     }
 
     public CompressionMetadata open(long overrideLength)

src/java/org/apache/cassandra/io/sstable/SimpleSSTableMultiWriter.java

@@ -132,7 +132,7 @@ public class SimpleSSTableMultiWriter implements SSTableMultiWriter
                                             .setMetadataCollector(metadataCollector)
                                             .setSerializationHeader(header)
                                             .addDefaultComponents(indexGroups)
-                                            .addFlushObserversForSecondaryIndexes(indexGroups, lifecycleNewTracker, metadata.get())
+                                            .setSecondaryIndexGroups(indexGroups)
                                             .build(lifecycleNewTracker, owner);
         return new SimpleSSTableMultiWriter(writer, lifecycleNewTracker);
     }

src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java

@@ -18,6 +18,7 @@
 
 package org.apache.cassandra.io.sstable.format;
 
+import java.io.IOError;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -30,6 +31,9 @@ import java.util.function.Supplier;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.db.SerializationHeader;
@@ -51,7 +55,6 @@ import org.apache.cassandra.io.sstable.metadata.MetadataComponent;
 import org.apache.cassandra.io.sstable.metadata.MetadataType;
 import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
 import org.apache.cassandra.io.util.MmappedRegionsCache;
-import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.utils.Throwables;
 import org.apache.cassandra.utils.TimeUUID;
 import org.apache.cassandra.utils.concurrent.Transactional;
@@ -65,6 +68,8 @@ import static com.google.common.base.Preconditions.checkNotNull;
  */
 public abstract class SSTableWriter extends SSTable implements Transactional
 {
+    private final static Logger logger = LoggerFactory.getLogger(SSTableWriter.class);
+
     protected long repairedAt;
     protected TimeUUID pendingRepair;
     protected boolean isTransient;
@@ -72,7 +77,7 @@ public abstract class SSTableWriter extends SSTable implements Transactional
     protected final long keyCount;
     protected final MetadataCollector metadataCollector;
     protected final SerializationHeader header;
-    protected final Collection<SSTableFlushObserver> observers;
+    protected final List<SSTableFlushObserver> observers;
     protected final MmappedRegionsCache mmappedRegionsCache;
     protected final TransactionalProxy txnProxy = txnProxy();
     protected final LifecycleNewTracker lifecycleNewTracker;
@@ -88,7 +93,7 @@ public abstract class SSTableWriter extends SSTable implements Transactional
     protected SSTableWriter(Builder<?, ?> builder, LifecycleNewTracker lifecycleNewTracker, SSTable.Owner owner)
     {
         super(builder, owner);
-        checkNotNull(builder.getFlushObservers());
+        checkNotNull(builder.getIndexGroups());
         checkNotNull(builder.getMetadataCollector());
         checkNotNull(builder.getSerializationHeader());
 
@@ -98,11 +103,60 @@ public abstract class SSTableWriter extends SSTable implements Transactional
         this.isTransient = builder.isTransientSSTable();
         this.metadataCollector = builder.getMetadataCollector();
         this.header = builder.getSerializationHeader();
-        this.observers = builder.getFlushObservers();
         this.mmappedRegionsCache = builder.getMmappedRegionsCache();
         this.lifecycleNewTracker = lifecycleNewTracker;
 
+        // We need to ensure that no sstable components exist before the lifecycle transaction starts tracking it.
+        // Otherwise, it means that we either want to overwrite some existing sstable, which is not allowed, or some
+        // sstable files were created before the sstable is registered in the lifecycle transaction, which may lead
+        // to a race such that the sstable is listed as completed due to the lack of the transaction file before
+        // anything is actually written to it.
+        Set<Component> existingComponents = Sets.filter(components, c -> descriptor.fileFor(c).exists());
+        assert existingComponents.isEmpty() : String.format("Cannot create a new SSTable in directory %s as component files %s already exist there",
+                                                            descriptor.directory,
+                                                            existingComponents);
+
         lifecycleNewTracker.trackNew(this);
+
+        try
+        {
+            ArrayList<SSTableFlushObserver> observers = new ArrayList<>();
+            this.observers = Collections.unmodifiableList(observers);
+            for (Index.Group group : builder.getIndexGroups())
+            {
+                SSTableFlushObserver observer = group.getFlushObserver(descriptor, lifecycleNewTracker, metadata.getLocal());
+                if (observer != null)
+                {
+                    observer.begin();
+                    observers.add(observer);
+                }
+            }
+        }
+        catch (RuntimeException | IOError ex)
+        {
+            handleConstructionFailure(ex);
+            throw ex;
+        }
+    }
+
+    /**
+     * Constructors of subclasses, if they open any resources, should wrap that in a try-catch block and call this
+     * method in the 'catch' section after closing any resources opened in the constructor. This method would remove
+     * the sstable from the transaction and delete the orphaned components, if any were created during the construction.
+     * The caught exception should be then rethrown so the {@link Builder} can handle it and close any resources opened
+     * implicitly by the builder.
+     * <p>
+     * See {@link SortedTableWriter#SortedTableWriter(SortedTableWriter.Builder, LifecycleNewTracker, Owner)} as of CASSANDRA-18737.
+     *
+     * @param ex the exception thrown during the construction
+     */
+    protected void handleConstructionFailure(Throwable ex)
+    {
+        logger.warn("Failed to open " + descriptor + " for writing", ex);
+        for (int i = observers.size()-1; i >= 0; i--)
+            observers.get(i).abort(ex);
+        descriptor.getFormat().deleteOrphanedComponents(descriptor, components);
+        lifecycleNewTracker.untrackNew(this);
     }
 
     @Override
@@ -382,7 +436,7 @@ public abstract class SSTableWriter extends SSTable implements Transactional
         private TimeUUID pendingRepair;
         private boolean transientSSTable;
         private SerializationHeader serializationHeader;
-        private Collection<SSTableFlushObserver> flushObservers;
+        private List<Index.Group> indexGroups;
 
         public B setMetadataCollector(MetadataCollector metadataCollector)
         {
@@ -420,12 +474,6 @@ public abstract class SSTableWriter extends SSTable implements Transactional
             return (B) this;
         }
 
-        public B setFlushObservers(Collection<SSTableFlushObserver> flushObservers)
-        {
-            this.flushObservers = ImmutableList.copyOf(flushObservers);
-            return (B) this;
-        }
-
         public B addDefaultComponents(Collection<Index.Group> indexGroups)
         {
             checkNotNull(getTableMetadataRef());
@@ -460,26 +508,11 @@ public abstract class SSTableWriter extends SSTable implements Transactional
             return components;
         }
 
-        public B addFlushObserversForSecondaryIndexes(Collection<Index.Group> indexGroups, LifecycleNewTracker tracker, TableMetadata metadata)
+        public B setSecondaryIndexGroups(Collection<Index.Group> indexGroups)
         {
-            if (indexGroups == null)
+            checkNotNull(indexGroups);
-                return (B) this;
+            this.indexGroups = ImmutableList.copyOf(indexGroups);
-
+            return (B) this;
-            Collection<SSTableFlushObserver> current = this.flushObservers != null ? this.flushObservers : Collections.emptyList();
-            List<SSTableFlushObserver> observers = new ArrayList<>(indexGroups.size() + current.size());
-            observers.addAll(current);
-
-            for (Index.Group group : indexGroups)
-            {
-                SSTableFlushObserver observer = group.getFlushObserver(descriptor, tracker, metadata);
-                if (observer != null)
-                {
-                    observer.begin();
-                    observers.add(observer);
-                }
-            }
-
-            return setFlushObservers(observers);
         }
 
         public MetadataCollector getMetadataCollector()
@@ -512,9 +545,9 @@ public abstract class SSTableWriter extends SSTable implements Transactional
             return serializationHeader;
         }
 
-        public Collection<SSTableFlushObserver> getFlushObservers()
+        public List<Index.Group> getIndexGroups()
         {
-            return flushObservers;
+            return indexGroups == null ? Collections.emptyList() : indexGroups;
         }
 
         public abstract MmappedRegionsCache getMmappedRegionsCache();

src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java

@@ -23,7 +23,9 @@ import java.nio.BufferOverflowException;
 import java.util.Collection;
 import java.util.Set;
 import java.util.function.Consumer;
+import java.util.function.Supplier;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -71,27 +73,48 @@ import static com.google.common.base.Preconditions.checkNotNull;
 /**
  * A generic implementation of a writer which assumes the existence of some partition index and bloom filter.
  */
-public abstract class SortedTableWriter<P extends SortedTablePartitionWriter> extends SSTableWriter
+public abstract class SortedTableWriter<P extends SortedTablePartitionWriter, I extends SortedTableWriter.AbstractIndexWriter> extends SSTableWriter
 {
     private final static Logger logger = LoggerFactory.getLogger(SortedTableWriter.class);
 
     // TODO dataWriter is not needed to be directly accessible - we can access everything we need for the dataWriter
     //   from a partition writer
     protected final SequentialWriter dataWriter;
+    protected final I indexWriter;
     protected final P partitionWriter;
     private final FileHandle.Builder dataFileBuilder = new FileHandle.Builder(descriptor.fileFor(Components.DATA));
     private DecoratedKey lastWrittenKey;
     private DataPosition dataMark;
     private long lastEarlyOpenLength;
 
-    public SortedTableWriter(Builder<P, ?, ?> builder, LifecycleNewTracker lifecycleNewTracker, SSTable.Owner owner)
+    public SortedTableWriter(Builder<P, I, ?, ?> builder, LifecycleNewTracker lifecycleNewTracker, SSTable.Owner owner)
     {
         super(builder, lifecycleNewTracker, owner);
-        checkNotNull(builder.getDataWriter());
-        checkNotNull(builder.getPartitionWriter());
 
-        this.dataWriter = builder.getDataWriter();
+        SequentialWriter dataWriter = null;
-        this.partitionWriter = builder.getPartitionWriter();
+        I indexWriter = null;
+        P partitionWriter = null;
+        try
+        {
+            dataWriter = builder.openDataWriter();
+            checkNotNull(dataWriter);
+
+            indexWriter = builder.openIndexWriter(dataWriter);
+            checkNotNull(indexWriter);
+
+            partitionWriter = builder.openPartitionWriter(dataWriter, indexWriter);
+            checkNotNull(partitionWriter);
+
+            this.dataWriter = dataWriter;
+            this.indexWriter = indexWriter;
+            this.partitionWriter = partitionWriter;
+        }
+        catch (RuntimeException | Error ex)
+        {
+            Throwables.closeNonNullAndAddSuppressed(ex, partitionWriter, indexWriter, dataWriter);
+            handleConstructionFailure(ex);
+            throw ex;
+        }
     }
 
     /**
@@ -264,6 +287,7 @@ public abstract class SortedTableWriter<P extends SortedTablePartitionWriter> ex
     public void mark()
     {
         dataMark = dataWriter.mark();
+        indexWriter.mark();
     }
 
     @Override
@@ -271,6 +295,29 @@ public abstract class SortedTableWriter<P extends SortedTablePartitionWriter> ex
     {
         dataWriter.resetAndTruncate(dataMark);
         partitionWriter.reset();
+        indexWriter.resetAndTruncate();
+    }
+
+    @Override
+    protected SSTableWriter.TransactionalProxy txnProxy()
+    {
+        return new TransactionalProxy(() -> FBUtilities.immutableListWithFilteredNulls(indexWriter, dataWriter));
+    }
+
+    protected class TransactionalProxy extends SSTableWriter.TransactionalProxy
+    {
+        public TransactionalProxy(Supplier<ImmutableList<Transactional>> transactionals)
+        {
+            super(transactionals);
+        }
+
+        @Override
+        protected Throwable doPostCleanup(Throwable accumulate)
+        {
+            accumulate = Throwables.close(accumulate, partitionWriter);
+            accumulate = super.doPostCleanup(accumulate);
+            return accumulate;
+        }
     }
 
     @Override
@@ -385,7 +432,7 @@ public abstract class SortedTableWriter<P extends SortedTablePartitionWriter> ex
 
         protected final IFilter bf;
 
-        protected AbstractIndexWriter(Builder<?, ?, ?> b)
+        protected AbstractIndexWriter(Builder<?, ?, ?, ?> b)
         {
             this.descriptor = b.descriptor;
             this.metadata = b.getTableMetadataRef();
@@ -409,6 +456,10 @@ public abstract class SortedTableWriter<P extends SortedTablePartitionWriter> ex
             }
         }
 
+        public abstract void mark();
+
+        public abstract void resetAndTruncate();
+
         protected void doPrepare()
         {
             flushBf();
@@ -428,8 +479,9 @@ public abstract class SortedTableWriter<P extends SortedTablePartitionWriter> ex
     }
 
     public abstract static class Builder<P extends SortedTablePartitionWriter,
-                                                  W extends SortedTableWriter<P>,
+                                        I extends AbstractIndexWriter,
-                                                  B extends Builder<P, W, B>> extends SSTableWriter.Builder<W, B>
+                                        W extends SortedTableWriter<P, I>,
+                                        B extends Builder<P, I, W, B>> extends SSTableWriter.Builder<W, B>
     {
 
         public Builder(Descriptor descriptor)
@@ -450,8 +502,10 @@ public abstract class SortedTableWriter<P extends SortedTablePartitionWriter> ex
             return (B) this;
         }
 
-        public abstract SequentialWriter getDataWriter();
+        protected abstract SequentialWriter openDataWriter();
 
-        public abstract P getPartitionWriter();
+        protected abstract I openIndexWriter(SequentialWriter dataWriter);
+
+        protected abstract P openPartitionWriter(SequentialWriter dataWriter, I indexWriter);
     }
 }

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

@@ -24,15 +24,14 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.function.Consumer;
 
-import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableSet;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.db.DeletionTime;
+import org.apache.cassandra.db.compaction.OperationType;
 import org.apache.cassandra.db.lifecycle.ILifecycleTransaction;
 import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
 import org.apache.cassandra.index.Index;
@@ -44,7 +43,6 @@ import org.apache.cassandra.io.sstable.SSTable;
 import org.apache.cassandra.io.sstable.format.DataComponent;
 import org.apache.cassandra.io.sstable.format.IndexComponent;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
-import org.apache.cassandra.io.sstable.format.SSTableWriter;
 import org.apache.cassandra.io.sstable.format.SortedTableWriter;
 import org.apache.cassandra.io.sstable.format.big.BigFormat.Components;
 import org.apache.cassandra.io.sstable.indexsummary.IndexSummary;
@@ -59,20 +57,19 @@ import org.apache.cassandra.io.util.SequentialWriter;
 import org.apache.cassandra.service.CacheService;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.EstimatedHistogram;
-import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.IFilter;
 import org.apache.cassandra.utils.JVMStabilityInspector;
 import org.apache.cassandra.utils.Throwables;
 
 import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
 import static org.apache.cassandra.io.util.FileHandle.Builder.NO_LENGTH_OVERRIDE;
 import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
 
-public class BigTableWriter extends SortedTableWriter<BigFormatPartitionWriter>
+public class BigTableWriter extends SortedTableWriter<BigFormatPartitionWriter, BigTableWriter.IndexWriter>
 {
     private static final Logger logger = LoggerFactory.getLogger(BigTableWriter.class);
 
-    private final IndexWriter indexWriter;
     private final RowIndexEntry.IndexSerializer rowIndexEntrySerializer;
     private final Map<DecoratedKey, AbstractRowIndexEntry> cachedKeys = new HashMap<>();
     private final boolean shouldMigrateKeyCache;
@@ -80,30 +77,15 @@ public class BigTableWriter extends SortedTableWriter<BigFormatPartitionWriter>
     public BigTableWriter(Builder builder, LifecycleNewTracker lifecycleNewTracker, SSTable.Owner owner)
     {
         super(builder, lifecycleNewTracker, owner);
-        checkNotNull(builder.getRowIndexEntrySerializer());
-        checkNotNull(builder.getIndexWriter());
 
         this.rowIndexEntrySerializer = builder.getRowIndexEntrySerializer();
-        this.indexWriter = builder.getIndexWriter();
+        checkNotNull(this.rowIndexEntrySerializer);
+
         this.shouldMigrateKeyCache = DatabaseDescriptor.shouldMigrateKeycacheOnCompaction()
                                      && lifecycleNewTracker instanceof ILifecycleTransaction
                                      && !((ILifecycleTransaction) lifecycleNewTracker).isOffline();
     }
 
-    @Override
-    public void mark()
-    {
-        super.mark();
-        indexWriter.mark();
-    }
-
-    @Override
-    public void resetAndTruncate()
-    {
-        super.resetAndTruncate();
-        indexWriter.resetAndTruncate();
-    }
-
     @Override
     protected void onStartPartition(DecoratedKey key)
     {
@@ -145,7 +127,7 @@ public class BigTableWriter extends SortedTableWriter<BigFormatPartitionWriter>
         return entry;
     }
 
-    @SuppressWarnings({"resource", "RedundantSuppression"})
+    @SuppressWarnings({ "resource", "RedundantSuppression" })
     private BigTableReader openInternal(IndexSummaryBuilder.ReadableBoundary boundary, SSTableReader.OpenReason openReason)
     {
         assert boundary == null || (boundary.indexLength > 0 && boundary.dataLength > 0);
@@ -251,16 +233,10 @@ public class BigTableWriter extends SortedTableWriter<BigFormatPartitionWriter>
         return openInternal(null, openReason);
     }
 
-    @Override
-    protected SSTableWriter.TransactionalProxy txnProxy()
-    {
-        return new SSTableWriter.TransactionalProxy(() -> FBUtilities.immutableListWithFilteredNulls(indexWriter, dataWriter));
-    }
-
     /**
      * Encapsulates writing the index and filter for an SSTable. The state of this object is not valid until it has been closed.
      */
-    static class IndexWriter extends SortedTableWriter.AbstractIndexWriter
+    protected static class IndexWriter extends SortedTableWriter.AbstractIndexWriter
     {
         private final RowIndexEntry.IndexSerializer rowIndexEntrySerializer;
 
@@ -271,7 +247,7 @@ public class BigTableWriter extends SortedTableWriter<BigFormatPartitionWriter>
         private DecoratedKey first;
         private DecoratedKey last;
 
-        protected IndexWriter(Builder b)
+        protected IndexWriter(Builder b, SequentialWriter dataWriter)
         {
             super(b);
             this.rowIndexEntrySerializer = b.getRowIndexEntrySerializer();
@@ -279,10 +255,8 @@ public class BigTableWriter extends SortedTableWriter<BigFormatPartitionWriter>
             builder = IndexComponent.fileBuilder(Components.PRIMARY_INDEX, b).withMmappedRegionsCache(b.getMmappedRegionsCache());
             summary = new IndexSummaryBuilder(b.getKeyCount(), b.getTableMetadataRef().getLocal().params.minIndexInterval, Downsampling.BASE_SAMPLING_LEVEL);
             // register listeners to be alerted when the data files are flushed
-            writer.setPostFlushListener(() -> summary.markIndexSynced(writer.getLastFlushOffset()));
+            writer.setPostFlushListener(summary::markIndexSynced);
-            @SuppressWarnings({"resource", "RedundantSuppression"})
+            dataWriter.setPostFlushListener(summary::markDataSynced);
-            SequentialWriter dataWriter = b.getDataWriter();
-            dataWriter.setPostFlushListener(() -> summary.markDataSynced(dataWriter.getLastFlushOffset()));
         }
 
         // finds the last (-offset) decorated key that can be guaranteed to occur fully in the flushed portion of the index file
@@ -316,11 +290,13 @@ public class BigTableWriter extends SortedTableWriter<BigFormatPartitionWriter>
             summary.maybeAddEntry(key, indexStart, indexEnd, dataEnd);
         }
 
+        @Override
         public void mark()
         {
             mark = writer.mark();
         }
 
+        @Override
         public void resetAndTruncate()
         {
             // we can't un-set the bloom filter addition, but extra keys in there are harmless.
@@ -372,13 +348,17 @@ public class BigTableWriter extends SortedTableWriter<BigFormatPartitionWriter>
         }
     }
 
-    public static class Builder extends SortedTableWriter.Builder<BigFormatPartitionWriter, BigTableWriter, Builder>
+    public static class Builder extends SortedTableWriter.Builder<BigFormatPartitionWriter, IndexWriter, BigTableWriter, Builder>
     {
         private RowIndexEntry.IndexSerializer rowIndexEntrySerializer;
-        private IndexWriter indexWriter;
-        private SequentialWriter dataWriter;
-        private BigFormatPartitionWriter partitionWriter;
         private MmappedRegionsCache mmappedRegionsCache;
+        private OperationType operationType;
+
+        // Writers are expected to be opened only once during construction of the sstable. The following flags are used
+        // to ensure that.
+        private boolean indexWriterOpened;
+        private boolean dataWriterOpened;
+        private boolean partitionWriterOpened;
 
         public Builder(Descriptor descriptor)
         {
@@ -405,30 +385,51 @@ public class BigTableWriter extends SortedTableWriter<BigFormatPartitionWriter>
         }
 
         @Override
-        public SequentialWriter getDataWriter()
+        protected SequentialWriter openDataWriter()
         {
-            return ensuringInBuildInternalContext(dataWriter);
+            checkState(!dataWriterOpened, "Data writer has been already opened.");
+
+            SequentialWriter dataWriter = DataComponent.buildWriter(descriptor,
+                                                                    getTableMetadataRef().getLocal(),
+                                                                    getIOOptions().writerOptions,
+                                                                    getMetadataCollector(),
+                                                                    ensuringInBuildInternalContext(operationType),
+                                                                    getIOOptions().flushCompression);
+            this.dataWriterOpened = true;
+            return dataWriter;
         }
 
         @Override
-        public BigFormatPartitionWriter getPartitionWriter()
+        protected IndexWriter openIndexWriter(SequentialWriter dataWriter)
         {
-            return ensuringInBuildInternalContext(partitionWriter);
+            checkNotNull(dataWriter);
+            checkState(!indexWriterOpened, "Index writer has been already opened.");
+
+            IndexWriter indexWriter = new IndexWriter(this, dataWriter);
+            this.indexWriterOpened = true;
+            return indexWriter;
         }
 
-        public RowIndexEntry.IndexSerializer getRowIndexEntrySerializer()
+        @Override
+        protected BigFormatPartitionWriter openPartitionWriter(SequentialWriter dataWriter, IndexWriter indexWriter)
+        {
+            checkNotNull(dataWriter);
+            checkNotNull(indexWriter);
+            checkState(!partitionWriterOpened, "Partition writer has been already opened.");
+
+            BigFormatPartitionWriter partitionWriter = new BigFormatPartitionWriter(getSerializationHeader(), dataWriter, descriptor.version, getRowIndexEntrySerializer().indexInfoSerializer());
+            this.partitionWriterOpened = true;
+            return partitionWriter;
+        }
+
+        RowIndexEntry.IndexSerializer getRowIndexEntrySerializer()
         {
             return ensuringInBuildInternalContext(rowIndexEntrySerializer);
         }
 
-        public IndexWriter getIndexWriter()
-        {
-            return ensuringInBuildInternalContext(indexWriter);
-        }
-
         private <T> T ensuringInBuildInternalContext(T value)
         {
-            Preconditions.checkState(value != null, "This getter can be used only during the lifetime of the sstable constructor. Do not use it directly.");
+            checkState(value != null, "The requested resource has not been initialized yet.");
             return value;
         }
 
@@ -437,30 +438,24 @@ public class BigTableWriter extends SortedTableWriter<BigFormatPartitionWriter>
         {
             try
             {
-                mmappedRegionsCache = new MmappedRegionsCache();
+                this.operationType = lifecycleNewTracker.opType();
-                rowIndexEntrySerializer = new RowIndexEntry.Serializer(descriptor.version, getSerializationHeader(), owner != null ? owner.getMetrics() : null);
+                this.mmappedRegionsCache = new MmappedRegionsCache();
-                dataWriter = DataComponent.buildWriter(descriptor,
+                this.rowIndexEntrySerializer = new RowIndexEntry.Serializer(descriptor.version, getSerializationHeader(), owner != null ? owner.getMetrics() : null);
-                                                       getTableMetadataRef().getLocal(),
-                                                       getIOOptions().writerOptions,
-                                                       getMetadataCollector(),
-                                                       lifecycleNewTracker.opType(),
-                                                       getIOOptions().flushCompression);
-                indexWriter = new IndexWriter(this);
-                partitionWriter = new BigFormatPartitionWriter(getSerializationHeader(), dataWriter, descriptor.version, rowIndexEntrySerializer.indexInfoSerializer());
                 return new BigTableWriter(this, lifecycleNewTracker, owner);
             }
             catch (RuntimeException | Error ex)
             {
-                Throwables.closeAndAddSuppressed(ex, partitionWriter, indexWriter, dataWriter, mmappedRegionsCache);
+                Throwables.closeNonNullAndAddSuppressed(ex, mmappedRegionsCache);
                 throw ex;
             }
             finally
             {
-                rowIndexEntrySerializer = null;
+                this.rowIndexEntrySerializer = null;
-                indexWriter = null;
+                this.mmappedRegionsCache = null;
-                dataWriter = null;
+                this.operationType = null;
-                partitionWriter = null;
+                this.partitionWriterOpened = false;
-                mmappedRegionsCache = null;
+                this.indexWriterOpened = false;
+                this.dataWriterOpened = false;
             }
         }
     }

src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableWriter.java

@@ -23,14 +23,13 @@ import java.util.function.Consumer;
 import java.util.function.Supplier;
 
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.db.DeletionTime;
+import org.apache.cassandra.db.compaction.OperationType;
 import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
 import org.apache.cassandra.index.Index;
 import org.apache.cassandra.io.FSReadError;
@@ -50,44 +49,25 @@ import org.apache.cassandra.io.util.MmappedRegionsCache;
 import org.apache.cassandra.io.util.SequentialWriter;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.Clock;
-import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.IFilter;
 import org.apache.cassandra.utils.JVMStabilityInspector;
 import org.apache.cassandra.utils.Throwables;
-import org.apache.cassandra.utils.concurrent.Transactional;
 
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
 import static org.apache.cassandra.io.util.FileHandle.Builder.NO_LENGTH_OVERRIDE;
 
 /**
  * Writes SSTables in BTI format (see {@link BtiFormat}), which can be read by {@link BtiTableReader}.
  */
 @VisibleForTesting
-public class BtiTableWriter extends SortedTableWriter<BtiFormatPartitionWriter>
+public class BtiTableWriter extends SortedTableWriter<BtiFormatPartitionWriter, BtiTableWriter.IndexWriter>
 {
     private static final Logger logger = LoggerFactory.getLogger(BtiTableWriter.class);
 
-    private final BtiFormatPartitionWriter partitionWriter;
-    private final IndexWriter iwriter;
-
     public BtiTableWriter(Builder builder, LifecycleNewTracker lifecycleNewTracker, SSTable.Owner owner)
     {
         super(builder, lifecycleNewTracker, owner);
-        this.iwriter = builder.getIndexWriter();
-        this.partitionWriter = builder.getPartitionWriter();
-    }
-
-    @Override
-    public void mark()
-    {
-        super.mark();
-        iwriter.mark();
-    }
-
-    @Override
-    public void resetAndTruncate()
-    {
-        super.resetAndTruncate();
-        iwriter.resetAndTruncate();
     }
 
     @Override
@@ -97,11 +77,11 @@ public class BtiTableWriter extends SortedTableWriter<BtiFormatPartitionWriter>
                                                      finishResult,
                                                      partitionLevelDeletion,
                                                      partitionWriter.getRowIndexBlockCount());
-        iwriter.append(key, entry);
+        indexWriter.append(key, entry);
         return entry;
     }
 
-    @SuppressWarnings({"resource", "RedundantSuppression"})
+    @SuppressWarnings({ "resource", "RedundantSuppression" })
     private BtiTableReader openInternal(OpenReason openReason, boolean isFinal, Supplier<PartitionIndex> partitionIndexSupplier)
     {
         IFilter filter = null;
@@ -118,9 +98,9 @@ public class BtiTableWriter extends SortedTableWriter<BtiFormatPartitionWriter>
             builder.setStatsMetadata(statsMetadata());
 
             partitionIndex = partitionIndexSupplier.get();
-            rowIndexFile = iwriter.rowIndexFHBuilder.complete();
+            rowIndexFile = indexWriter.rowIndexFHBuilder.complete();
             dataFile = openDataFile(isFinal ? NO_LENGTH_OVERRIDE : dataWriter.getLastFlushOffset(), builder.getStatsMetadata());
-            filter = iwriter.getFilterCopy();
+            filter = indexWriter.getFilterCopy();
 
             return builder.setPartitionIndex(partitionIndex)
                           .setFirst(partitionIndex.firstKey())
@@ -142,9 +122,9 @@ public class BtiTableWriter extends SortedTableWriter<BtiFormatPartitionWriter>
     public void openEarly(Consumer<SSTableReader> callWhenReady)
     {
         long dataLength = dataWriter.position();
-        iwriter.buildPartial(dataLength, partitionIndex ->
+        indexWriter.buildPartial(dataLength, partitionIndex ->
         {
-            iwriter.rowIndexFHBuilder.withLengthOverride(iwriter.rowIndexWriter.getLastFlushOffset());
+            indexWriter.rowIndexFHBuilder.withLengthOverride(indexWriter.rowIndexWriter.getLastFlushOffset());
             BtiTableReader reader = openInternal(OpenReason.EARLY, false, () -> partitionIndex);
             callWhenReady.accept(reader);
         });
@@ -154,10 +134,10 @@ public class BtiTableWriter extends SortedTableWriter<BtiFormatPartitionWriter>
     public SSTableReader openFinalEarly()
     {
         // we must ensure the data is completely flushed to disk
-        iwriter.complete(); // This will be called by completedPartitionIndex() below too, but we want it done now to
+        indexWriter.complete(); // This will be called by completedPartitionIndex() below too, but we want it done now to
         // ensure outstanding openEarly actions are not triggered.
         dataWriter.sync();
-        iwriter.rowIndexWriter.sync();
+        indexWriter.rowIndexWriter.sync();
         // Note: Nothing must be written to any of the files after this point, as the chunk cache could pick up and
         // retain a partially-written page.
 
@@ -165,42 +145,20 @@ public class BtiTableWriter extends SortedTableWriter<BtiFormatPartitionWriter>
     }
 
     @Override
-    @SuppressWarnings({"resource", "RedundantSuppression"})
+    @SuppressWarnings({ "resource", "RedundantSuppression" })
     protected SSTableReader openFinal(OpenReason openReason)
     {
 
         if (maxDataAge < 0)
             maxDataAge = Clock.Global.currentTimeMillis();
 
-        return openInternal(openReason, true, iwriter::completedPartitionIndex);
+        return openInternal(openReason, true, indexWriter::completedPartitionIndex);
-    }
-
-    @Override
-    protected TransactionalProxy txnProxy()
-    {
-        return new TransactionalProxy(() -> FBUtilities.immutableListWithFilteredNulls(iwriter, dataWriter));
-    }
-
-    private class TransactionalProxy extends SortedTableWriter<BtiFormatPartitionWriter>.TransactionalProxy
-    {
-        public TransactionalProxy(Supplier<ImmutableList<Transactional>> transactionals)
-        {
-            super(transactionals);
-        }
-
-        @Override
-        protected Throwable doPostCleanup(Throwable accumulate)
-        {
-            accumulate = Throwables.close(accumulate, partitionWriter);
-            accumulate = super.doPostCleanup(accumulate);
-            return accumulate;
-        }
     }
 
     /**
      * Encapsulates writing the index and filter for an SSTable. The state of this object is not valid until it has been closed.
      */
-    static class IndexWriter extends SortedTableWriter.AbstractIndexWriter
+    protected static class IndexWriter extends SortedTableWriter.AbstractIndexWriter
     {
         final SequentialWriter rowIndexWriter;
         private final FileHandle.Builder rowIndexFHBuilder;
@@ -211,7 +169,7 @@ public class BtiTableWriter extends SortedTableWriter<BtiFormatPartitionWriter>
         private DataPosition riMark;
         private DataPosition piMark;
 
-        IndexWriter(Builder b)
+        IndexWriter(Builder b, SequentialWriter dataWriter)
         {
             super(b);
             rowIndexWriter = new SequentialWriter(descriptor.fileFor(Components.ROW_INDEX), b.getIOOptions().writerOptions);
@@ -220,11 +178,9 @@ public class BtiTableWriter extends SortedTableWriter<BtiFormatPartitionWriter>
             partitionIndexFHBuilder = IndexComponent.fileBuilder(Components.PARTITION_INDEX, b).withMmappedRegionsCache(b.getMmappedRegionsCache());
             partitionIndex = new PartitionIndexBuilder(partitionIndexWriter, partitionIndexFHBuilder);
             // register listeners to be alerted when the data files are flushed
-            partitionIndexWriter.setPostFlushListener(() -> partitionIndex.markPartitionIndexSynced(partitionIndexWriter.getLastFlushOffset()));
+            partitionIndexWriter.setPostFlushListener(partitionIndex::markPartitionIndexSynced);
-            rowIndexWriter.setPostFlushListener(() -> partitionIndex.markRowIndexSynced(rowIndexWriter.getLastFlushOffset()));
+            rowIndexWriter.setPostFlushListener(partitionIndex::markRowIndexSynced);
-            @SuppressWarnings({"resource", "RedundantSuppression"})
+            dataWriter.setPostFlushListener(partitionIndex::markDataSynced);
-            SequentialWriter dataWriter = b.getDataWriter();
-            dataWriter.setPostFlushListener(() -> partitionIndex.markDataSynced(dataWriter.getLastFlushOffset()));
         }
 
         public long append(DecoratedKey key, AbstractRowIndexEntry indexEntry) throws IOException
@@ -336,50 +292,20 @@ public class BtiTableWriter extends SortedTableWriter<BtiFormatPartitionWriter>
         }
     }
 
-    public static class Builder extends SortedTableWriter.Builder<BtiFormatPartitionWriter, BtiTableWriter, Builder>
+    public static class Builder extends SortedTableWriter.Builder<BtiFormatPartitionWriter, IndexWriter, BtiTableWriter, Builder>
     {
-        private SequentialWriter dataWriter;
-        private BtiFormatPartitionWriter partitionWriter;
-        private IndexWriter indexWriter;
         private MmappedRegionsCache mmappedRegionsCache;
+        private OperationType operationType;
+
+        private boolean dataWriterOpened;
+        private boolean partitionWriterOpened;
+        private boolean indexWriterOpened;
 
         public Builder(Descriptor descriptor)
         {
             super(descriptor);
         }
 
-        // The following getters for the resources opened by buildInternal method can be only used during the lifetime of
-        // that method - that is, during the construction of the sstable.
-
-        @Override
-        public MmappedRegionsCache getMmappedRegionsCache()
-        {
-            return ensuringInBuildInternalContext(mmappedRegionsCache);
-        }
-
-        @Override
-        public SequentialWriter getDataWriter()
-        {
-            return ensuringInBuildInternalContext(dataWriter);
-        }
-
-        @Override
-        public BtiFormatPartitionWriter getPartitionWriter()
-        {
-            return ensuringInBuildInternalContext(partitionWriter);
-        }
-
-        public IndexWriter getIndexWriter()
-        {
-            return ensuringInBuildInternalContext(indexWriter);
-        }
-
-        private <T> T ensuringInBuildInternalContext(T value)
-        {
-            Preconditions.checkState(value != null, "This getter can be used only during the lifetime of the sstable constructor. Do not use it directly.");
-            return value;
-        }
-
         @Override
         public Builder addDefaultComponents(Collection<Index.Group> indexGroups)
         {
@@ -390,40 +316,82 @@ public class BtiTableWriter extends SortedTableWriter<BtiFormatPartitionWriter>
             return this;
         }
 
+        // The following getters for the resources opened by buildInternal method can be only used during the lifetime of
+        // that method - that is, during the construction of the sstable.
+
+        @Override
+        public MmappedRegionsCache getMmappedRegionsCache()
+        {
+            return ensuringInBuildInternalContext(mmappedRegionsCache);
+        }
+
+        @Override
+        protected SequentialWriter openDataWriter()
+        {
+            checkState(!dataWriterOpened, "Data writer has been already opened.");
+
+            return DataComponent.buildWriter(descriptor,
+                                             getTableMetadataRef().getLocal(),
+                                             getIOOptions().writerOptions,
+                                             getMetadataCollector(),
+                                             ensuringInBuildInternalContext(operationType),
+                                             getIOOptions().flushCompression);
+        }
+
+        @Override
+        protected IndexWriter openIndexWriter(SequentialWriter dataWriter)
+        {
+            checkNotNull(dataWriter);
+            checkState(!indexWriterOpened, "Index writer has been already opened.");
+
+            IndexWriter indexWriter = new IndexWriter(this, dataWriter);
+            indexWriterOpened = true;
+            return indexWriter;
+        }
+
+        @Override
+        protected BtiFormatPartitionWriter openPartitionWriter(SequentialWriter dataWriter, IndexWriter indexWriter)
+        {
+            checkNotNull(dataWriter);
+            checkNotNull(indexWriter);
+            checkState(!partitionWriterOpened, "Partition writer has been already opened.");
+
+            BtiFormatPartitionWriter partitionWriter = new BtiFormatPartitionWriter(getSerializationHeader(),
+                                                                                    getTableMetadataRef().getLocal().comparator,
+                                                                                    dataWriter,
+                                                                                    indexWriter.rowIndexWriter,
+                                                                                    descriptor.version);
+            partitionWriterOpened = true;
+            return partitionWriter;
+        }
+
+        private <T> T ensuringInBuildInternalContext(T value)
+        {
+            checkState(value != null, "The requested resource has not been initialized yet.");
+            return value;
+        }
+
         @Override
         protected BtiTableWriter buildInternal(LifecycleNewTracker lifecycleNewTracker, Owner owner)
         {
             try
             {
-                mmappedRegionsCache = new MmappedRegionsCache();
+                this.mmappedRegionsCache = new MmappedRegionsCache();
-                dataWriter = DataComponent.buildWriter(descriptor,
+                this.operationType = lifecycleNewTracker.opType();
-                                                       getTableMetadataRef().getLocal(),
-                                                       getIOOptions().writerOptions,
-                                                       getMetadataCollector(),
-                                                       lifecycleNewTracker.opType(),
-                                                       getIOOptions().flushCompression);
-
-                indexWriter = new IndexWriter(this);
-                partitionWriter = new BtiFormatPartitionWriter(getSerializationHeader(),
-                                                               getTableMetadataRef().getLocal().comparator,
-                                                               dataWriter,
-                                                               indexWriter.rowIndexWriter,
-                                                               descriptor.version);
-
 
                 return new BtiTableWriter(this, lifecycleNewTracker, owner);
             }
             catch (RuntimeException | Error ex)
             {
-                Throwables.closeAndAddSuppressed(ex, partitionWriter, indexWriter, dataWriter, mmappedRegionsCache);
+                Throwables.closeAndAddSuppressed(ex, mmappedRegionsCache);
                 throw ex;
             }
             finally
             {
-                partitionWriter = null;
-                indexWriter = null;
-                dataWriter = null;
                 mmappedRegionsCache = null;
+                partitionWriterOpened = false;
+                indexWriterOpened = false;
+                dataWriterOpened = false;
             }
         }
     }

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

@@ -21,6 +21,7 @@ import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.nio.file.StandardOpenOption;
+import java.util.function.LongConsumer;
 
 import org.apache.cassandra.io.FSReadError;
 import org.apache.cassandra.io.FSWriteError;
@@ -60,7 +61,7 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr
 
     protected long lastFlushOffset;
 
-    protected Runnable runPostFlush;
+    protected LongConsumer runPostFlush;
 
     private final TransactionalProxy txnProxy = txnProxy();
 
@@ -229,7 +230,7 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr
         resetBuffer();
     }
 
-    public void setPostFlushListener(Runnable runPostFlush)
+    public void setPostFlushListener(LongConsumer runPostFlush)
     {
         assert this.runPostFlush == null;
         this.runPostFlush = runPostFlush;
@@ -252,7 +253,7 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr
             throw new FSWriteError(e, getPath());
         }
         if (runPostFlush != null)
-            runPostFlush.run();
+            runPostFlush.accept(getLastFlushOffset());
     }
 
     @Override

test/unit/org/apache/cassandra/db/SerializationHeaderTest.java

@@ -100,7 +100,6 @@ public class SerializationHeaderTest
                                                              .setTableMetadataRef(TableMetadataRef.forOfflineTools(schema))
                                                              .setKeyCount(1)
                                                              .setSerializationHeader(header)
-                                                             .setFlushObservers(Collections.emptyList())
                                                              .setMetadataCollector(new MetadataCollector(schema.comparator))
                                                              .addDefaultComponents(Collections.emptySet())
                                                              .build(txn, null))

test/unit/org/apache/cassandra/db/lifecycle/RealTransactionsTest.java

@@ -163,7 +163,7 @@ public class RealTransactionsTest extends SchemaLoader
                 rewriter.switchWriter(desc.getFormat().getWriterFactory().builder(desc)
                                           .setTableMetadataRef(metadata)
                                           .setSerializationHeader(SerializationHeader.make(cfs.metadata(), txn.originals()))
-                                          .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), txn, metadata.get())
+                                          .setSecondaryIndexGroups(cfs.indexManager.listIndexGroups())
                                           .setMetadataCollector(new MetadataCollector(cfs.metadata().comparator))
                                           .addDefaultComponents(cfs.indexManager.listIndexGroups())
                                           .build(txn, cfs));

test/unit/org/apache/cassandra/io/sstable/SSTableFlushObserverTest.java

@@ -23,12 +23,16 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Iterator;
+import java.util.List;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
 
 import com.google.common.collect.ArrayListMultimap;
 import com.google.common.collect.Multimap;
 import org.apache.commons.lang3.tuple.ImmutableTriple;
 import org.apache.commons.lang3.tuple.Triple;
 import org.junit.Assert;
+import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
@@ -50,6 +54,7 @@ import org.apache.cassandra.db.rows.Cell;
 import org.apache.cassandra.db.rows.EncodingStats;
 import org.apache.cassandra.db.rows.Row;
 import org.apache.cassandra.db.rows.Unfiltered;
+import org.apache.cassandra.index.Index;
 import org.apache.cassandra.io.FSWriteError;
 import org.apache.cassandra.io.sstable.format.SSTableFormat;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
@@ -60,102 +65,169 @@ import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.schema.TableMetadataRef;
+import org.apache.cassandra.utils.concurrent.Transactional.AbstractTransactional.State;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatRuntimeException;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 public class SSTableFlushObserverTest
 {
+
+    private TableMetadata cfm;
+    private File directory;
+    private SSTableFormat<?, ?> sstableFormat;
+    private static Supplier<SSTableId> idGen;
+
     @BeforeClass
     public static void initDD()
     {
         DatabaseDescriptor.daemonInitialization();
         CommitLog.instance.start();
+        idGen = SSTableIdFactory.instance.defaultBuilder().generator(Stream.empty());
     }
 
     private static final String KS_NAME = "test";
     private static final String CF_NAME = "flush_observer";
 
-    @Test
+    @Before
-    public void testFlushObserver() throws Exception
+    public void setUp() throws Exception
     {
-        TableMetadata cfm =
+        cfm = TableMetadata.builder(KS_NAME, CF_NAME)
-        TableMetadata.builder(KS_NAME, CF_NAME)
                      .addPartitionKeyColumn("id", UTF8Type.instance)
                      .addRegularColumn("first_name", UTF8Type.instance)
                      .addRegularColumn("age", Int32Type.instance)
                      .addRegularColumn("height", LongType.instance)
                      .build();
-
-        LifecycleTransaction transaction = LifecycleTransaction.offline(OperationType.COMPACTION);
-        FlushObserver observer = new FlushObserver();
-
         String sstableDirectory = DatabaseDescriptor.getAllDataFileLocations()[0];
-        File directory = new File(sstableDirectory + File.pathSeparator() + KS_NAME + File.pathSeparator() + CF_NAME);
+        directory = new File(sstableDirectory + File.pathSeparator() + KS_NAME + File.pathSeparator() + CF_NAME);
         directory.deleteOnExit();
 
         if (!directory.exists() && !directory.tryCreateDirectories())
             throw new FSWriteError(new IOException("failed to create tmp directory"), directory.absolutePath());
 
-        SSTableFormat<?, ?> sstableFormat = DatabaseDescriptor.getSelectedSSTableFormat();
+        sstableFormat = DatabaseDescriptor.getSelectedSSTableFormat();
-        Descriptor descriptor = new Descriptor(sstableFormat.getLatestVersion(),
+    }
-                                               directory,
-                                               cfm.keyspace,
-                                               cfm.name,
-                                               new SequenceBasedSSTableId(0));
 
-        SSTableWriter writer = descriptor.getFormat().getWriterFactory().builder(descriptor)
+    @Test
-                                         .setKeyCount(10)
+    public void testFlushObserver() throws Exception
-                                         .setTableMetadataRef(TableMetadataRef.forOfflineTools(cfm))
+    {
-                                         .setMetadataCollector(new MetadataCollector(cfm.comparator).sstableLevel(0))
+        try (LifecycleTransaction transaction = LifecycleTransaction.offline(OperationType.COMPACTION))
-                                         .setSerializationHeader(new SerializationHeader(true, cfm, cfm.regularAndStaticColumns(), EncodingStats.NO_STATS))
-                                         .setFlushObservers(Collections.singletonList(observer))
-                                         .addDefaultComponents(Collections.emptySet())
-                                         .build(transaction, null);
-
-        SSTableReader reader;
-        Multimap<ByteBuffer, Cell<?>> expected = ArrayListMultimap.create();
-
-        try
         {
-            final long now = System.currentTimeMillis();
+            FlushObserver observer = new FlushObserver();
 
-            ByteBuffer key = UTF8Type.instance.fromString("key1");
+            Descriptor descriptor = new Descriptor(sstableFormat.getLatestVersion(),
-            expected.putAll(key, Arrays.asList(BufferCell.live(getColumn(cfm, "age"), now, Int32Type.instance.decompose(27)),
+                                                   directory,
-                                               BufferCell.live(getColumn(cfm, "first_name"), now, UTF8Type.instance.fromString("jack")),
+                                                   cfm.keyspace,
-                                               BufferCell.live(getColumn(cfm, "height"), now, LongType.instance.decompose(183L))));
+                                                   cfm.name,
+                                                   idGen.get());
+            Index.Group indexGroup = mock(Index.Group.class);
+            when(indexGroup.getFlushObserver(any(), any(), any())).thenReturn(observer);
+            SSTableWriter.Builder<?, ?> writerBuilder = descriptor.getFormat().getWriterFactory().builder(descriptor)
+                                                                  .setKeyCount(10)
+                                                                  .setTableMetadataRef(TableMetadataRef.forOfflineTools(cfm))
+                                                                  .setMetadataCollector(new MetadataCollector(cfm.comparator).sstableLevel(0))
+                                                                  .setSerializationHeader(new SerializationHeader(true, cfm, cfm.regularAndStaticColumns(), EncodingStats.NO_STATS))
+                                                                  .setSecondaryIndexGroups(Collections.singleton(indexGroup))
+                                                                  .addDefaultComponents(Collections.emptySet());
+            assertThat(observer.beginCalled).isFalse();
+            assertThat(observer.isComplete).isFalse();
 
-            writer.append(new RowIterator(cfm, key.duplicate(), Collections.singletonList(buildRow(expected.get(key)))));
+            SSTableWriter writer = writerBuilder.build(transaction, null);
 
-            key = UTF8Type.instance.fromString("key2");
+            assertThat(observer.beginCalled).isTrue();
-            expected.putAll(key, Arrays.asList(BufferCell.live(getColumn(cfm, "age"), now, Int32Type.instance.decompose(30)),
+            assertThat(observer.isComplete).isFalse();
-                                               BufferCell.live(getColumn(cfm, "first_name"), now, UTF8Type.instance.fromString("jim")),
-                                               BufferCell.live(getColumn(cfm, "height"), now, LongType.instance.decompose(180L))));
 
-            writer.append(new RowIterator(cfm, key, Collections.singletonList(buildRow(expected.get(key)))));
+            SSTableReader reader;
+            Multimap<ByteBuffer, Cell<?>> expected = ArrayListMultimap.create();
 
-            key = UTF8Type.instance.fromString("key3");
+            try
-            expected.putAll(key, Arrays.asList(BufferCell.live(getColumn(cfm, "age"), now, Int32Type.instance.decompose(30)),
+            {
-                                               BufferCell.live(getColumn(cfm, "first_name"), now, UTF8Type.instance.fromString("ken")),
+                final long now = System.currentTimeMillis();
-                                               BufferCell.live(getColumn(cfm, "height"), now, LongType.instance.decompose(178L))));
 
-            writer.append(new RowIterator(cfm, key, Collections.singletonList(buildRow(expected.get(key)))));
+                ByteBuffer key = UTF8Type.instance.fromString("key1");
+                expected.putAll(key, Arrays.asList(BufferCell.live(getColumn(cfm, "age"), now, Int32Type.instance.decompose(27)),
+                                                   BufferCell.live(getColumn(cfm, "first_name"), now, UTF8Type.instance.fromString("jack")),
+                                                   BufferCell.live(getColumn(cfm, "height"), now, LongType.instance.decompose(183L))));
 
-            reader = writer.finish(true);
+                writer.append(new RowIterator(cfm, key.duplicate(), Collections.singletonList(buildRow(expected.get(key)))));
+
+                key = UTF8Type.instance.fromString("key2");
+                expected.putAll(key, Arrays.asList(BufferCell.live(getColumn(cfm, "age"), now, Int32Type.instance.decompose(30)),
+                                                   BufferCell.live(getColumn(cfm, "first_name"), now, UTF8Type.instance.fromString("jim")),
+                                                   BufferCell.live(getColumn(cfm, "height"), now, LongType.instance.decompose(180L))));
+
+                writer.append(new RowIterator(cfm, key, Collections.singletonList(buildRow(expected.get(key)))));
+
+                key = UTF8Type.instance.fromString("key3");
+                expected.putAll(key, Arrays.asList(BufferCell.live(getColumn(cfm, "age"), now, Int32Type.instance.decompose(30)),
+                                                   BufferCell.live(getColumn(cfm, "first_name"), now, UTF8Type.instance.fromString("ken")),
+                                                   BufferCell.live(getColumn(cfm, "height"), now, LongType.instance.decompose(178L))));
+
+                writer.append(new RowIterator(cfm, key, Collections.singletonList(buildRow(expected.get(key)))));
+
+                reader = writer.finish(true);
+            }
+            finally
+            {
+                FileUtils.closeQuietly(writer);
+            }
+
+            Assert.assertTrue(observer.isComplete);
+            Assert.assertEquals(expected.size(), observer.rows.size());
+
+            for (Triple<ByteBuffer, Long, Long> e : observer.rows.keySet())
+            {
+                ByteBuffer key = e.getLeft();
+                long indexPosition = e.getRight();
+
+                DecoratedKey indexKey = reader.keyAtPositionFromSecondaryIndex(indexPosition);
+                Assert.assertEquals(0, UTF8Type.instance.compare(key, indexKey.getKey()));
+                Assert.assertEquals(expected.get(key), observer.rows.get(e));
+            }
         }
-        finally
+    }
+
+    @Test
+    public void testFailedInitialization() throws Exception
+    {
+        try (LifecycleTransaction transaction = LifecycleTransaction.offline(OperationType.COMPACTION))
         {
-            FileUtils.closeQuietly(writer);
+            FlushObserver observer1 = new FlushObserver();
-        }
+            FlushObserver observer2 = new FlushObserver();
+            Index.Group indexGroup1 = mock(Index.Group.class);
+            Index.Group indexGroup2 = mock(Index.Group.class);
+            when(indexGroup1.getFlushObserver(any(), any(), any())).thenReturn(observer1);
+            when(indexGroup2.getFlushObserver(any(), any(), any())).thenReturn(observer2);
+            observer2.failOnBegin = true;
 
-        Assert.assertTrue(observer.isComplete);
+            Descriptor descriptor = new Descriptor(sstableFormat.getLatestVersion(),
-        Assert.assertEquals(expected.size(), observer.rows.size());
+                                                   directory,
+                                                   cfm.keyspace,
+                                                   cfm.name,
+                                                   idGen.get());
+            assertThatRuntimeException().isThrownBy(() -> descriptor.getFormat().getWriterFactory().builder(descriptor)
+                                                                    .setKeyCount(10)
+                                                                    .setTableMetadataRef(TableMetadataRef.forOfflineTools(cfm))
+                                                                    .setMetadataCollector(new MetadataCollector(cfm.comparator).sstableLevel(0))
+                                                                    .setSerializationHeader(new SerializationHeader(true, cfm, cfm.regularAndStaticColumns(), EncodingStats.NO_STATS))
+                                                                    .setSecondaryIndexGroups(List.of(indexGroup1, indexGroup2))
+                                                                    .addDefaultComponents(Collections.emptySet())
+                                                                    .build(transaction, null)
+            ).withMessage("Failed to initialize");
 
-        for (Triple<ByteBuffer, Long, Long> e : observer.rows.keySet())
+            assertThat(observer1.beginCalled).isTrue();
-        {
+            assertThat(observer1.isComplete).isFalse();
-            ByteBuffer key = e.getLeft();
+            assertThat(observer1.abortCalled).isTrue();
-            long indexPosition = e.getRight();
 
-            DecoratedKey indexKey = reader.keyAtPositionFromSecondaryIndex(indexPosition);
+            assertThat(observer2.beginCalled).isTrue();
-            Assert.assertEquals(0, UTF8Type.instance.compare(key, indexKey.getKey()));
+            assertThat(observer2.isComplete).isFalse();
-            Assert.assertEquals(expected.get(key), observer.rows.get(e));
+            assertThat(observer2.abortCalled).isFalse();
+
+            assertThat(transaction.state()).isEqualTo(State.IN_PROGRESS);
+            assertThat(transaction.originals()).isEmpty();
         }
     }
 
@@ -190,10 +262,16 @@ public class SSTableFlushObserverTest
 
         private Triple<ByteBuffer, Long, Long> currentKey;
         private boolean isComplete;
+        private boolean beginCalled;
+        private boolean failOnBegin;
+        private boolean abortCalled;
 
         @Override
         public void begin()
         {
+            beginCalled = true;
+            if (failOnBegin)
+                throw new RuntimeException("Failed to initialize");
         }
 
         @Override
@@ -220,6 +298,12 @@ public class SSTableFlushObserverTest
         {
             staticRow.forEach((c) -> staticRows.put(currentKey, (Cell<?>) c));
         }
+
+        @Override
+        public void abort(Throwable accumulate)
+        {
+            abortCalled = true;
+        }
     }
 
     private static Row buildRow(Collection<Cell<?>> cells)

test/unit/org/apache/cassandra/io/sstable/SSTableWriterTestBase.java

@@ -163,7 +163,7 @@ public class SSTableWriterTestBase extends SchemaLoader
                    .setPendingRepair(pendingRepair)
                    .setTransientSSTable(isTransient)
                    .setSerializationHeader(new SerializationHeader(true, cfs.metadata(), cfs.metadata().regularAndStaticColumns(), EncodingStats.NO_STATS))
-                   .addFlushObserversForSecondaryIndexes(cfs.indexManager.listIndexGroups(), txn, cfs.metadata.get())
+                   .setSecondaryIndexGroups(cfs.indexManager.listIndexGroups())
                    .setMetadataCollector(new MetadataCollector(cfs.metadata().comparator))
                    .addDefaultComponents(cfs.indexManager.listIndexGroups())
                    .build(txn, cfs);

test/unit/org/apache/cassandra/io/sstable/ScrubTest.java

@@ -839,7 +839,6 @@ public class ScrubTest
                                          .setTableMetadataRef(cfs.metadata)
                                          .setMetadataCollector(collector)
                                          .setSerializationHeader(header)
-                                         .setFlushObservers(Collections.emptyList())
                                          .addDefaultComponents(Collections.emptySet())
                                          .build(txn, cfs);
 

test/unit/org/apache/cassandra/io/sstable/format/bti/PartitionIndexTest.java

@@ -429,7 +429,7 @@ public class PartitionIndexTest
                  PartitionIndexBuilder builder = new PartitionIndexBuilder(writer, fhBuilder)
             )
             {
-                writer.setPostFlushListener(() -> builder.markPartitionIndexSynced(writer.getLastFlushOffset()));
+                writer.setPostFlushListener(builder::markPartitionIndexSynced);
                 for (int i = 0; i < COUNT; i++)
                 {
                     DecoratedKey key = generateRandomLengthKey();
@@ -506,7 +506,7 @@ public class PartitionIndexTest
                     for (i = 0; i < 3; ++i)
                         builder.addEntry(list.get(i), i);
 
-                    writer.setPostFlushListener(() -> builder.markPartitionIndexSynced(writer.getLastFlushOffset()));
+                    writer.setPostFlushListener(builder::markPartitionIndexSynced);
                     AtomicInteger callCount = new AtomicInteger();
 
                     final int addedSize = i;
@@ -669,7 +669,7 @@ public class PartitionIndexTest
                  PartitionIndexBuilder builder = new PartitionIndexBuilder(writer, fhBuilder)
             )
             {
-                writer.setPostFlushListener(() -> builder.markPartitionIndexSynced(writer.getLastFlushOffset()));
+                writer.setPostFlushListener(builder::markPartitionIndexSynced);
                 for (int i = 0; i < COUNT; i++)
                 {
                     DecoratedKey key = generateRandomKey();
@@ -708,7 +708,7 @@ public class PartitionIndexTest
              PartitionIndexBuilder builder = new PartitionIndexBuilder(writer, fhBuilder)
         )
         {
-            writer.setPostFlushListener(() -> builder.markPartitionIndexSynced(writer.getLastFlushOffset()));
+            writer.setPostFlushListener(builder::markPartitionIndexSynced);
             for (int i = 0; i < 1000; i++)
             {
                 DecoratedKey key = generateRandomKey();