CNDB-14123: allows intercepting issues opening flushed sstables (#1742)

For riptano/cndb#14123, we want to be able to
catch issues happening during the opening of just-flushed sstables to
use shallow sstables. This commit enable this by adding a method to
`StorageHandler` that is called on such issue, and allow to provide a
"replacement" `SSTableReader` instance.

Author: pcmanus

src/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -1457,11 +1457,11 @@ public Collection<SSTableReader> flushMemtable(ColumnFamilyStore cfs, Memtable m
                     // This can throw on remote storage, e.g. if a file cannot be uploaded
                     txn.prepareToCommit();
 
-                    // Open the underlying readers, the one that will be returne below by `finished()`.
+                    // Open the underlying readers, the one that will be returned below by `finished()`.
                     // Currently needs to be called before commit, because committing will close a certain number
                     // of resources used by the writers which are accessed to open the readers.
                     for (SSTableMultiWriter writer : flushResults)
-                        writer.openResult();
+                        writer.openResult(storageHandler);
                 }
                 catch (Throwable t)
                 {
@@ -2579,7 +2579,7 @@ public void writeAndAddMemtableRanges(UUID repairSessionID,
         {
             try
             {
-                Collection<SSTableReader> sstables = memtableContent.finish(true);
+                Collection<SSTableReader> sstables = memtableContent.finish(true, storageHandler);
                 try (Refs sstableReferences = Refs.ref(sstables))
                 {
                     // This moves all references to placeIntoRefs, clearing sstableReferences

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

@@ -375,7 +375,8 @@ public List<SSTableReader>  scrub()
                 {
                     for (Partition partition : outOfOrder)
                         inOrderWriter.append(partition.unfilteredIterator());
-                    newInOrderSstable = inOrderWriter.finish(-1, sstable.maxDataAge, true);
+                    inOrderWriter.setMaxDataAge(sstable.maxDataAge);
+                    newInOrderSstable = inOrderWriter.finish(true, null);
                 }
                 transaction.update(newInOrderSstable, false);
                 finished.add(newInOrderSstable);

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

@@ -21,6 +21,8 @@
 import java.util.List;
 import java.util.UUID;
 
+import javax.annotation.Nullable;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -35,6 +37,7 @@
 import org.apache.cassandra.index.Index;
 import org.apache.cassandra.io.sstable.Descriptor;
 import org.apache.cassandra.io.sstable.SSTableMultiWriter;
+import org.apache.cassandra.io.sstable.StorageHandler;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.sstable.format.SSTableWriter;
 import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
@@ -143,27 +146,14 @@ public void append(UnfilteredRowIterator partition)
     }
 
     @Override
-    public Collection<SSTableReader> finish(long repairedAt, long maxDataAge, boolean openResult)
-    {
-        List<SSTableReader> sstables = new ArrayList<>(writers.length);
-        for (SSTableWriter writer : writers)
-            if (writer != null)
-            {
-                boundaries.applyTokenSpaceCoverage(writer);
-                sstables.add(writer.finish(repairedAt, maxDataAge, openResult));
-            }
-        return sstables;
-    }
-
-    @Override
-    public Collection<SSTableReader> finish(boolean openResult)
+    public Collection<SSTableReader> finish(boolean openResult, @Nullable StorageHandler storageHandler)
     {
         List<SSTableReader> sstables = new ArrayList<>(writers.length);
         for (SSTableWriter writer : writers)
             if (writer != null)
             {
                 boundaries.applyTokenSpaceCoverage(writer);
-                sstables.add(writer.finish(openResult));
+                sstables.add(writer.finish(openResult, storageHandler));
             }
         return sstables;
     }
@@ -179,11 +169,11 @@ public Collection<SSTableReader> finished()
     }
 
     @Override
-    public void openResult()
+    public void openResult(@Nullable StorageHandler storageHandler)
     {
         for (SSTableWriter writer : writers)
             if (writer != null)
-                writer.openResult();
+                writer.openResult(storageHandler);
     }
 
     @Override

src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java

@@ -104,7 +104,7 @@ public synchronized void received(IncomingStream stream)
         SSTableMultiWriter sstable = file.getSSTable();
         try
         {
-            finished = sstable.finish(true);
+            finished = sstable.finish(true, cfs.getStorageHandler());
         }
         catch (Throwable t)
         {

src/java/org/apache/cassandra/index/sai/disk/format/DefaultIndexComponentDiscovery.java

@@ -25,9 +25,14 @@
 public class DefaultIndexComponentDiscovery extends IndexComponentDiscovery
 {
     @Override
-    public SSTableIndexComponentsState discoverComponents(SSTableReader sstable) {
-        Descriptor descriptor = sstable.getDescriptor();
+    public SSTableIndexComponentsState discoverComponents(SSTableReader sstable)
+    {
+        return discoverComponents(sstable.getDescriptor(), sstable.metadata());
+    }
 
+    @Override
+    public SSTableIndexComponentsState discoverComponents(Descriptor descriptor, TableMetadata metadata)
+    {
         // Older versions might not have all components in the TOC, we should not trust it (fix for CNDB-13582):
         if (descriptor.version.getVersion().compareTo("ca") < 0)
            return discoverComponentsFromDiskFallback(descriptor);

src/java/org/apache/cassandra/index/sai/disk/format/IndexComponentDiscovery.java

@@ -77,14 +77,34 @@ public static IndexComponentDiscovery instance()
      * be used for each group of components (per-sstable and per-index).
      *
      * @param sstable the sstable reader for which to discover components.
-     * @return the discovered {@link ComponentsBuildId} to use for both per-sstable and each per-index components. The
-     * returned build IDs should usually correspond to existing index components on disk but this is not a strong
+     * @return the discovered {@link ComponentsBuildId} to use for both per-sstable and each per-index component. The
+     * returned build IDs should usually correspond to existing index components on disk, but this is not a strong
      * asumption: if some group of components corresponding to the returned build ID has no completion marker or is
-     * missing files, the group will not be usuable (and the corresponding index/indexes will not be usable) but this
+     * missing files, the group will not be usuable (and the corresponding index/indexes will not be usable), but this
      * should be handled "gracefully" by callers.
      */
     public abstract SSTableIndexComponentsState discoverComponents(SSTableReader sstable);
 
+    /**
+     * Returns the set of groups of SAI components that should be used for the provided sstable.
+     * <p>
+     * Note that "discovery" in this method only means finding out the "build ID" (version and generation) that should
+     * be used for each group of components (per-sstable and per-index).
+     * <p>
+     * Please note that the {@link #discoverComponents(SSTableReader)} method should be prefered when a
+     * {@link SSTableReader} exists as some implementations may be more optimal or use additional checks when provided
+     * a reader.
+     *
+     * @param descriptor the descriptor of the sstable for which to discover components.
+     * @param metadata the metadata of the table the sstable belongs to.
+     * @return the discovered {@link ComponentsBuildId} to use for both per-sstable and each per-index component. The
+     * returned build IDs should usually correspond to existing index components on disk, but this is not a strong
+     * asumption: if some group of components corresponding to the returned build ID has no completion marker or is
+     * missing files, the group will not be usuable (and the corresponding index/indexes will not be usable), but this
+     * should be handled "gracefully" by callers.
+     */
+    public abstract SSTableIndexComponentsState discoverComponents(Descriptor descriptor, TableMetadata metadata);
+
     protected static IndexComponentType completionMarker(@Nullable String name)
     {
         return name == null ? IndexComponentType.GROUP_COMPLETION_MARKER : IndexComponentType.COLUMN_COMPLETION_MARKER;

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

@@ -20,6 +20,8 @@
 
 import java.util.Collection;
 
+import javax.annotation.Nullable;
+
 import org.apache.cassandra.db.rows.UnfilteredRowIterator;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.schema.TableId;
@@ -36,17 +38,21 @@ public interface SSTableMultiWriter extends Transactional
      */
     void append(UnfilteredRowIterator partition);
 
-    Collection<SSTableReader> finish(long repairedAt, long maxDataAge, boolean openResult);
-    Collection<SSTableReader> finish(boolean openResult);
+    Collection<SSTableReader> finish(boolean openResult, @Nullable StorageHandler handler);
     Collection<SSTableReader> finished();
 
     /**
      * Opens the resulting sstables after writing has finished. If those readers need to be accessed, then this must
      * be called after `prepareToCommit` (so the writing is complete) but before `commit` (because committing closes
      * some of the resources used to create the underlying readers). When used, the readers can then be accessed by
      * calling `finished()`.
+     *
+     * @param storageHandler the underlying storage handler. This is used in case of a failure opening the
+     *                       SSTableReader(s) to call the `StorageHandler#onOpeningWrittenSSTableFailure` callback, which
+     *                       in some implementations may attempt to recover from the error. If `null`, the said callback
+     *                       will not be called on failure (and the exception will propagate).
      */
-    void openResult();
+    void openResult(@Nullable StorageHandler storageHandler);
 
     String getFilename();
     long getBytesWritten();

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

@@ -354,7 +354,7 @@ protected void doPrepare()
         {
             assert writer.getFilePointer() > 0;
             writer.setRepairedAt(repairedAt).prepareToCommit();
-            writer.openResult();
+            writer.openResult(null);
             SSTableReader reader = writer.finished();
             transaction.update(reader, false);
             preparedForCommit.add(reader);

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

@@ -260,7 +260,7 @@ public void run()
                     {
                         for (Map.Entry<DecoratedKey, PartitionUpdate.Builder> entry : b.entrySet())
                             writer.append(entry.getValue().build().unfilteredIterator());
-                        writer.finish(false);
+                        writer.finish(false, null);
                     }
                 }
                 catch (Throwable e)

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

@@ -81,7 +81,7 @@ public void close()
             if (update != null)
                 writePartition(update.build());
             if (writer != null)
-                writer.finish(false);
+                writer.finish(false, null);
         }
         catch (Throwable t)
         {