diff --git a/CHANGES.txt b/CHANGES.txt
index 4b04683842..1b325ab56e 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -200,6 +200,7 @@
  * Add the ability to disable bulk loading of SSTables (CASSANDRA-18781)
  * Clean up obsolete functions and simplify cql_version handling in cqlsh (CASSANDRA-18787)
 Merged from 5.0:
+ * Avoid availability gap between UP and queryability marking for already built SAI indexes on bounce (CASSANDRA-20732)
  * Make Commitlog flush data safely in Direct IO mode (CASSANDRA-20692)
  * Get SAI MemtableIndex refs before SSTableIndex refs at query time (CASSANDRA-20709)
  * Fix MAX_SEGMENT_SIZE < chunkSize in MmappedRegions::updateState (CASSANDRA-20636)
diff --git a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java
index 98a48b02e2..c17f9fb0a9 100644
--- a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java
+++ b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java
@@ -332,9 +332,30 @@ public class StorageAttachedIndex implements Index
     public Callable<?> getInitializationTask()
     {
         // New storage-attached indexes will be available for queries after on disk index data are built.
-        // Memtable data will be indexed via flushing triggered by schema change
-        // We only want to validate the index files if we are starting up
-        IndexValidation validation = StorageService.instance.isStarting() ? IndexValidation.HEADER_FOOTER : IndexValidation.NONE;
+        // Memtable data will be indexed via flushing triggered by schema change.
+        // We only want to validate the index files if we are starting up.
+        boolean isStarting = StorageService.instance.isStarting();
+        IndexValidation validation = isStarting ? IndexValidation.HEADER_FOOTER : IndexValidation.NONE;
+
+        // Only attempt to make the index queryable if we are starting up. Otherwise, if we create a new index on top
+        // of nothing but existing Memtable data (i.e. no SSTables), that data will temporarily be lost until flush.
+        if (isStarting)
+        {
+            StorageAttachedIndexGroup indexGroup = StorageAttachedIndexGroup.getIndexGroup(baseCfs);
+            assert indexGroup != null : "Index group does not exist for table " + baseCfs.keyspace + '.' + baseCfs.name;
+
+            Collection<SSTableReader> nonIndexed = findNonIndexedSSTables(baseCfs, indexGroup, validation);
+
+            if (nonIndexed.isEmpty())
+            {
+                // If the index is complete, mark it queryable and avoid an initial build:
+                baseCfs.indexManager.makeIndexQueryable(this, Status.BUILD_SUCCEEDED);
+                logger.debug(indexIdentifier.logMessage("Skipping initial build, as index is already queryable..."));
+                initBuildStarted = true;
+                return () -> ImmediateFuture.success(null);
+            }
+        }
+
         return () -> startInitialBuild(baseCfs, validation).get();
     }
 
@@ -847,15 +868,12 @@ public class StorageAttachedIndex implements Index
         // Force another flush to make sure on disk index is generated for memtable data before marking it queryable.
         // In the case of offline scrub, there are no live memtables.
         if (!baseCfs.getTracker().getView().liveMemtables.isEmpty())
-        {
             baseCfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.INDEX_BUILD_STARTED);
-        }
 
         // It is now safe to flush indexes directly from flushing Memtables.
         initBuildStarted = true;
 
         StorageAttachedIndexGroup indexGroup = StorageAttachedIndexGroup.getIndexGroup(baseCfs);
-
         assert indexGroup != null : "Index group does not exist for table " + baseCfs.keyspace + '.' + baseCfs.name;
 
         List<SSTableReader> nonIndexed = findNonIndexedSSTables(baseCfs, indexGroup, validation);
@@ -892,8 +910,7 @@ public class StorageAttachedIndex implements Index
             }
 
             StorageAttachedIndexGroup indexGroup = StorageAttachedIndexGroup.getIndexGroup(baseCfs);
-
-            assert indexGroup != null : "Index group does not exist for table";
+            assert indexGroup != null : "Index group does not exist for table " + baseCfs.keyspace + '.' + baseCfs.name;
 
             Collection<SSTableReader> nonIndexed = findNonIndexedSSTables(baseCfs, indexGroup, IndexValidation.HEADER_FOOTER);
 
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 0caf26a3a6..54f6c0674a 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -682,6 +682,12 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         initialized = true;
     }
 
+    @VisibleForTesting
+    public void unsafeSetUninitialized()
+    {
+        initialized = false;
+    }
+
     public boolean isInitialized()
     {
         return initialized;
diff --git a/test/unit/org/apache/cassandra/index/sai/cql/AllowFilteringTest.java b/test/unit/org/apache/cassandra/index/sai/cql/AllowFilteringTest.java
index 3ae0905185..db7491c2a1 100644
--- a/test/unit/org/apache/cassandra/index/sai/cql/AllowFilteringTest.java
+++ b/test/unit/org/apache/cassandra/index/sai/cql/AllowFilteringTest.java
@@ -18,6 +18,7 @@
 
 package org.apache.cassandra.index.sai.cql;
 
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 import org.apache.cassandra.cql3.restrictions.StatementRestrictions;
@@ -26,6 +27,7 @@ import org.apache.cassandra.index.sai.SAITester;
 import org.apache.cassandra.index.sai.StorageAttachedIndex;
 import org.apache.cassandra.inject.Injections;
 import org.apache.cassandra.inject.InvokePointBuilder;
+import org.apache.cassandra.service.StorageService;
 
 import static java.lang.String.format;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -36,6 +38,12 @@ import static org.junit.Assert.assertNotNull;
  */
 public class AllowFilteringTest extends SAITester
 {
+    @BeforeClass
+    public static void setup()
+    {
+        StorageService.instance.unsafeSetInitialized();
+    }
+
     @Test
     public void testAllowFilteringOnFirstClusteringKeyColumn() throws Throwable
     {
diff --git a/test/unit/org/apache/cassandra/index/sai/cql/EmptyStringLifecycleTest.java b/test/unit/org/apache/cassandra/index/sai/cql/EmptyStringLifecycleTest.java
index 7ace636e4f..c4ca10233b 100644
--- a/test/unit/org/apache/cassandra/index/sai/cql/EmptyStringLifecycleTest.java
+++ b/test/unit/org/apache/cassandra/index/sai/cql/EmptyStringLifecycleTest.java
@@ -18,13 +18,21 @@
 
 package org.apache.cassandra.index.sai.cql;
 
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 import org.apache.cassandra.cql3.UntypedResultSet;
 import org.apache.cassandra.index.sai.SAITester;
+import org.apache.cassandra.service.StorageService;
 
 public class EmptyStringLifecycleTest extends SAITester
 {
+    @BeforeClass
+    public static void setup()
+    {
+        StorageService.instance.unsafeSetInitialized();
+    }
+
     @Test
     public void testBeforeAndAfterFlush()
     {
diff --git a/test/unit/org/apache/cassandra/index/sai/cql/StorageAttachedIndexDDLTest.java b/test/unit/org/apache/cassandra/index/sai/cql/StorageAttachedIndexDDLTest.java
index cc2e2f4b13..2df5316309 100644
--- a/test/unit/org/apache/cassandra/index/sai/cql/StorageAttachedIndexDDLTest.java
+++ b/test/unit/org/apache/cassandra/index/sai/cql/StorageAttachedIndexDDLTest.java
@@ -76,6 +76,7 @@ import org.apache.cassandra.inject.InvokePointBuilder;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.utils.Throwables;
 import org.assertj.core.api.Assertions;
 import org.mockito.Mockito;
@@ -1311,6 +1312,40 @@ public class StorageAttachedIndexDDLTest extends SAITester
         assertTrue(verifyChecksum(numericIndexTermType, numericIndexIdentifier));
     }
 
+    @Test
+    public void shouldMarkQueryableInInitializationTask() throws Throwable
+    {
+        createTable(CREATE_TABLE_TEMPLATE);
+        disableCompaction(KEYSPACE);
+        IndexIdentifier idxIdentifier = createIndexIdentifier(createIndexAsync(String.format(CREATE_INDEX_TEMPLATE, "v1")));
+
+        // create 10 SSTables
+        for (int i = 0; i < 10; i++)
+        {
+            execute("INSERT INTO %s (id1, v1, v2) VALUES (?, ?, ?)", String.valueOf(i), i, String.valueOf(i));
+            flush();
+        }
+
+        ResultSet rows = executeNet("SELECT id1 FROM %s WHERE v1 >= 5");
+        assertEquals(5, rows.all().size());
+
+        // Make the index artificially non-queryable:
+        ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
+        Index index = cfs.indexManager.getIndexByName(idxIdentifier.indexName);
+        cfs.indexManager.makeIndexNonQueryable(index, Index.Status.BUILD_FAILED);
+
+        // Query should fail with the index in an artificially non-queryable state:
+        assertThatThrownBy(() -> executeNet("SELECT id1 FROM %s WHERE v1 >= 5")).isInstanceOf(ReadFailureException.class);
+
+        // Node must not yet be initialized for pre-emptive index validation to occur:
+        StorageService.instance.unsafeSetUninitialized();
+        // Simply getting the initialization task (and not running it) will validate and mark the index queryable again:
+        cfs.indexManager.buildIndex(index);
+        StorageService.instance.unsafeSetInitialized();
+        rows = executeNet("SELECT id1 FROM %s WHERE v1 >= 5");
+        assertEquals(5, rows.all().size());
+    }
+
     @Test
     public void shouldRejectQueriesWithCustomExpressions()
     {
diff --git a/test/unit/org/apache/cassandra/index/sai/functional/FailureTest.java b/test/unit/org/apache/cassandra/index/sai/functional/FailureTest.java
index 6907006e08..15d6add6c2 100644
--- a/test/unit/org/apache/cassandra/index/sai/functional/FailureTest.java
+++ b/test/unit/org/apache/cassandra/index/sai/functional/FailureTest.java
@@ -20,6 +20,8 @@
  */
 package org.apache.cassandra.index.sai.functional;
 
+import org.assertj.core.api.Assertions;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 import org.apache.cassandra.db.marshal.Int32Type;
@@ -30,12 +32,18 @@ import org.apache.cassandra.index.sai.utils.IndexIdentifier;
 import org.apache.cassandra.index.sai.utils.IndexTermType;
 import org.apache.cassandra.inject.Injection;
 import org.apache.cassandra.inject.Injections;
-import org.assertj.core.api.Assertions;
+import org.apache.cassandra.service.StorageService;
 
 import static org.junit.Assert.assertEquals;
 
 public class FailureTest extends SAITester
 {
+    @BeforeClass
+    public static void setup()
+    {
+        StorageService.instance.unsafeSetInitialized();
+    }
+
     @Test
     public void shouldMakeIndexNonQueryableOnSSTableContextFailureDuringFlush() throws Throwable
     {
diff --git a/test/unit/org/apache/cassandra/index/sai/virtual/IndexesSystemViewTest.java b/test/unit/org/apache/cassandra/index/sai/virtual/IndexesSystemViewTest.java
index bde94c8a75..59dd8a3850 100644
--- a/test/unit/org/apache/cassandra/index/sai/virtual/IndexesSystemViewTest.java
+++ b/test/unit/org/apache/cassandra/index/sai/virtual/IndexesSystemViewTest.java
@@ -29,6 +29,7 @@ import org.apache.cassandra.index.sai.StorageAttachedIndex;
 import org.apache.cassandra.inject.Injections;
 import org.apache.cassandra.inject.InvokePointBuilder;
 import org.apache.cassandra.schema.SchemaConstants;
+import org.apache.cassandra.service.StorageService;
 
 /**
  * Tests the virtual table exposing storage-attached column index metadata.
@@ -58,6 +59,8 @@ public class IndexesSystemViewTest extends SAITester
     public static void setup()
     {
         VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(SchemaConstants.VIRTUAL_VIEWS, ImmutableList.of(new ColumnIndexesSystemView(SchemaConstants.VIRTUAL_VIEWS))));
+
+        StorageService.instance.unsafeSetInitialized();
     }
 
     @Test