Fix ColumnFamilyStore.getIfExists handling of dropped keyspaces

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

@@ -3431,7 +3431,7 @@ public static ColumnFamilyStore getIfExists(TableId id)
         if (metadata == null)
             return null;
 
-        Keyspace keyspace = Keyspace.open(metadata.keyspace);
+        Keyspace keyspace = Keyspace.openIfExists(metadata.keyspace);
         if (keyspace == null)
             return null;
 
@@ -3449,7 +3449,7 @@ public static ColumnFamilyStore getIfExists(String ksName, String cfName)
         if (ksName == null || cfName == null)
             return null;
 
-        Keyspace keyspace = Keyspace.open(ksName);
+        Keyspace keyspace = Keyspace.openIfExists(ksName);
         if (keyspace == null)
             return null;
 

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

@@ -35,6 +35,8 @@
 import java.util.function.Supplier;
 import java.util.stream.Stream;
 
+import javax.annotation.Nullable;
+
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Iterables;
 import com.google.common.util.concurrent.RateLimiter;
@@ -149,7 +151,23 @@ public static void unsetInitialized()
         }
     }
 
-    public static Keyspace open(String keyspaceName)
+    /**
+     * Convenience function for when getting {@code null} if the keyspace does not exist is more convenient than a
+     * {@link UnknownKeyspaceException}.
+     */
+    public static @Nullable Keyspace openIfExists(String keyspaceName)
+    {
+        try
+        {
+            return open(keyspaceName);
+        }
+        catch (UnknownKeyspaceException e)
+        {
+            return null;
+        }
+    }
+
+    public static Keyspace open(String keyspaceName) throws UnknownKeyspaceException
     {
         assert initialized || SchemaConstants.isLocalSystemKeyspace(keyspaceName) : "Initialized: " + initialized;
         return open(keyspaceName, Schema.instance, true);

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

@@ -39,6 +39,8 @@
 import org.apache.cassandra.db.partitions.Partition;
 import org.apache.cassandra.db.partitions.PartitionUpdate;
 import org.apache.cassandra.index.transactions.UpdateTransaction;
+import org.apache.cassandra.schema.SchemaTestUtil;
+import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.utils.concurrent.OpOrder;
 import org.json.simple.JSONArray;
 import org.json.simple.JSONObject;
@@ -803,4 +805,24 @@ public void performSnapshot(String snapshotName)
             }
         };
     }
+
+    /**
+     * Ensures `getIfExists` returns `null` (and does not throw) after a keyspace is dropped.
+     */
+    @Test
+    public void testGetIfExistsAfterKeyspaceDrop()
+    {
+        String testKS = "getIfExistsAfterKeyspaceDrop";
+        String testTable = "test_table";
+        SchemaLoader.createKeyspace(testKS,
+                                    KeyspaceParams.simple(1),
+                                    SchemaLoader.standardCFMD(testKS, testTable));
+
+        TableMetadata metadata = ColumnFamilyStore.getIfExists(testKS, testTable).metadata.get();
+
+        SchemaTestUtil.announceKeyspaceDrop(testKS);
+
+        assertNull(ColumnFamilyStore.getIfExists(metadata.id));
+        assertNull(ColumnFamilyStore.getIfExists(testKS, testTable));
+    }
 }

test/unit/org/apache/cassandra/schema/SchemaTestUtil.java

@@ -97,7 +97,7 @@ public static void announceTableUpdate(TableMetadata updated)
         Schema.instance.transform(schema -> schema.withAddedOrUpdated(ksm.withSwapped(ksm.tables.withSwapped(updated))));
     }
 
-    static void announceKeyspaceDrop(String ksName)
+    public static void announceKeyspaceDrop(String ksName)
     {
         KeyspaceMetadata oldKsm = Schema.instance.getKeyspaceMetadata(ksName);
         if (oldKsm == null)