CNDB-14207: Don't mark the SAI non-queryable if flush fails

This commit changes handling SAI index flush failures.

A flush does not force the index into the non-queryable state anymore.
We can do this because after failure we rollback any partially flushed
index components, and we abort the sstable flush as well. Therefore,
both the failed-to-flush memtable and memtable indexes remain intact
and can still serve queries.

This change has several advantages:
- the flush failure could be temporary and the flush may still succeed
  the next time,
- even if the problem with flushing persists, reads will run fine
- if this is a node-local problem, the other nodes have a chance to
  take over; a failure of one node does not propagate to the rest of
  the cluster

Author: pkolaczk

src/java/org/apache/cassandra/index/sai/disk/StorageAttachedIndexWriter.java

@@ -303,7 +303,7 @@ public void abort(Throwable accumulator, boolean fromIndex)
 
         // For non-compaction, make any indexes involved in this transaction non-queryable, as they will likely not match the backing table.
         // For compaction: the compaction task should be aborted and new sstables will not be added to tracker
-        if (fromIndex && opType != OperationType.COMPACTION)
+        if (fromIndex && opType != OperationType.COMPACTION && opType != OperationType.FLUSH)
             indices.forEach(StorageAttachedIndex::makeIndexNonQueryable);
 
         for (PerIndexWriter perIndexWriter : perIndexWriters)

test/unit/org/apache/cassandra/index/sai/functional/FlushingTest.java

@@ -23,12 +23,19 @@
 import org.junit.Test;
 
 import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.exceptions.ReadFailureException;
 import org.apache.cassandra.db.marshal.Int32Type;
 import org.apache.cassandra.index.sai.IndexContext;
 import org.apache.cassandra.index.sai.SAITester;
 import org.apache.cassandra.index.sai.disk.v1.kdtree.NumericIndexWriter;
+import org.apache.cassandra.inject.ActionBuilder;
+import org.apache.cassandra.inject.Expression;
+import org.apache.cassandra.inject.Injection;
+import org.apache.cassandra.inject.Injections;
+import org.apache.cassandra.inject.InvokePointBuilder;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
 
 public class FlushingTest extends SAITester
 {
@@ -84,4 +91,38 @@ public void testFlushingOverwriteDelete() throws Throwable
 
         assertIndexFilesInToc(indexFiles());
     }
+
+    @Test
+    public void testMemtableIndexFlushFailure() throws Throwable
+    {
+        Injection failMemtableComplete = Injections.newCustom("FailMemtableIndexWriterComplete")
+                                                   .add(InvokePointBuilder.newInvokePoint()
+                                                                          .onClass("org.apache.cassandra.index.sai.disk.v1.MemtableIndexWriter")
+                                                                          .onMethod("complete", "com.google.common.base.Stopwatch")
+                                                   )
+                                                   .add(ActionBuilder.newActionBuilder().actions()
+                                                                     .doThrow(java.io.IOException.class, Expression.quote("Byteman-injected fault in MemtableIndexWriter.complete"))
+                                                   )
+                                                   .build();
+        Injections.inject(failMemtableComplete);
+
+        createTable(CREATE_TABLE_TEMPLATE);
+        createIndex(String.format(CREATE_INDEX_TEMPLATE, "v1"));
+
+        String pkValue = "key_bm_flush_fail";
+        int indexedValue = 456;
+
+        execute("INSERT INTO %s (id1, v1) VALUES (?, ?)", pkValue, indexedValue);
+
+        // The Byteman rule will cause the SAI part of the flush to fail
+        assertThrows(RuntimeException.class, this::flush);
+
+        // Assert that the index is still queryable for the inserted data despite the injected fault.
+        ResultSet indexQueryResults = executeNet("SELECT id1 FROM %s WHERE v1 = ?", indexedValue);
+        assertEquals("The index should be still usable despite flush failure", 1, indexQueryResults.all().size());
+
+        // Assert that the table is still queryable by primary key.
+        ResultSet pkQueryResults = executeNet("SELECT v1 FROM %s WHERE id1 = ?", pkValue);
+        assertEquals("Table should still be queryable by primary key", 1, pkQueryResults.all().size());
+    }
 }