diff --git a/src/java/org/apache/cassandra/index/sai/disk/StorageAttachedIndexWriter.java b/src/java/org/apache/cassandra/index/sai/disk/StorageAttachedIndexWriter.java
index 0d0171323456..e3e6a4e7b781 100644
--- a/src/java/org/apache/cassandra/index/sai/disk/StorageAttachedIndexWriter.java
+++ b/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)
diff --git a/test/unit/org/apache/cassandra/index/sai/functional/FlushingTest.java b/test/unit/org/apache/cassandra/index/sai/functional/FlushingTest.java
index 23861fb21099..0be793b0cf5d 100644
--- a/test/unit/org/apache/cassandra/index/sai/functional/FlushingTest.java
+++ b/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());
+    }
 }