HCD-237: UCS settings json file cleanup fails (#2145)

### What is the issue
UCS settings files are not dropped after the table gets dropped. Instead
they are supposed to be cleared after the node restart. The cleanup is
faulty though and it prevents the node from startup.

Root Cause:
The cleanupControllerConfig() method in CompactionManager attempts to
verify if a table exists by calling getColumnFamilyStore(). When the
table is dropped, this method throws IllegalArgumentException, which was
not being caught. The existing catch block only handled
NullPointerException (for missing keyspace).

### What does this PR fix and why was it fixed
Extended the exception handler to catch both NullPointerException and
IllegalArgumentException, allowing orphaned controller-config.JSON files
to be properly identified and deleted during node restart.

Author: szymon-miezal

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

@@ -267,12 +267,17 @@ public static void cleanupControllerConfig()
                             //table exists so keep the file
                             Schema.instance.getKeyspaceInstance(names[0]).getColumnFamilyStore(names[1]);
                         }
-                        catch (NullPointerException e)
+                        catch (NullPointerException | IllegalArgumentException e)
                         {
                             //table does not exist so delete the file
-                            logger.debug("Removing " + file + " because it does not correspond to an existing table");
+                            logger.debug("Removing {} because it does not correspond to an existing table", file);
                             file.delete();
                         }
+                        catch (Throwable e)
+                        {
+                            logger.error("Encountered an exception while cleaning up the orphaned compaction settings file ({}) for {}.{}", file, names[0], names[1]);
+                            throw e;
+                        }
                     }
                     else if (names.length == 3) // if keyspace/table names are long, we include table id as a 3rd component while the keyspace and table names are abbreviated
                     {
@@ -281,7 +286,7 @@ else if (names.length == 3) // if keyspace/table names are long, we include tabl
                         if (Schema.instance.getTableMetadata(tableId) == null)
                         {
                             //table does not exist so delete the file
-                            logger.debug("Removing " + file + " because it does not correspond to an existing table");
+                            logger.debug("Removing {} because it does not correspond to an existing table", file);
                             file.delete();
                         }
                     }

test/distributed/org/apache/cassandra/distributed/test/CompactionControllerConfigTest.java

@@ -19,6 +19,7 @@
 package org.apache.cassandra.distributed.test;
 
 import java.util.Arrays;
+import java.util.function.Consumer;
 
 import org.junit.Test;
 
@@ -31,6 +32,7 @@
 import org.apache.cassandra.db.compaction.unified.Controller;
 import org.apache.cassandra.db.compaction.unified.StaticController;
 import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
 import org.apache.cassandra.schema.TableMetadata;
 
 
@@ -40,6 +42,7 @@
 import static org.apache.cassandra.SchemaLoader.standardCFMD;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
 public class CompactionControllerConfigTest extends TestBaseImpl
@@ -228,6 +231,84 @@ public void testVectorControllerConfig() throws Throwable
         vectorControllerConfig(false);
     }
 
+    /**
+     * Test to reproduce the bug where orphaned controller-config.JSON files
+     * cause IllegalArgumentException during node restart.
+     *
+     * The bug occurs when:
+     * 1. A table with UCS compaction strategy is created
+     * 2. Data is written and flushed
+     * 3. The UCS config file is saved
+     * 4. Table is dropped (file is NOT deleted)
+     * 5. Node is restarted (cleanupControllerConfig() throws IllegalArgumentException)
+     */
+    @Test
+    public void testDropTableOrphanedControllerConfigFileCleanup() throws Throwable
+    {
+        testOrphanedControllerConfigFileCleanup(cluster -> cluster.schemaChange(withKeyspace("DROP TABLE test_ks.test_table;")));
+    }
+
+    /**
+     * Same as testDropTableOrphanedControllerConfigFileCleanup but for dropping keyspace
+     */
+    @Test
+    public void testDropKeyspaceOrphanedControllerConfigFileCleanup() throws Throwable
+    {
+        testOrphanedControllerConfigFileCleanup(cluster -> cluster.schemaChange(withKeyspace("DROP KEYSPACE test_ks;")));
+    }
+
+    private void testOrphanedControllerConfigFileCleanup(Consumer<Cluster> schemaRemover) throws Throwable
+    {
+        try (Cluster cluster = init(Cluster.build(1).start()))
+        {
+            // create keyspace and table with UCS compaction strategy
+            cluster.schemaChange(withKeyspace("CREATE KEYSPACE test_ks WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};"));
+            cluster.schemaChange(withKeyspace("CREATE TABLE test_ks.test_table (pk int PRIMARY KEY, v int) WITH compaction = " +
+                                              "{'class': 'UnifiedCompactionStrategy', 'adaptive': 'true'};"));
+
+            // insert data and flush to trigger controller-config.JSON creation
+            for (int i = 0; i < 100; i++)
+            {
+                cluster.coordinator(1).execute(withKeyspace("INSERT INTO test_ks.test_table (pk, v) VALUES (?, ?)"),
+                                               ConsistencyLevel.ONE, i, i);
+            }
+
+            cluster.get(1).flush(KEYSPACE);
+
+            // store the controller-config.JSON file and verify it exists
+            cluster.get(1).runOnInstance(() -> {
+                CompactionManager.storeControllerConfig();
+                TableMetadata metadata = TableMetadata.minimal("test_ks", "test_table");
+                assertTrue("Controller config file should exist after flush",
+                           Controller.getControllerConfigPath(metadata).exists());
+            });
+
+            // drop the schema - this should delete the controller-config.JSON file but currently doesn't
+            schemaRemover.accept(cluster);
+
+            // verify the orphaned file still exists
+            cluster.get(1).runOnInstance(() -> {
+                // This assertion will pass, showing the file is orphaned
+                TableMetadata metadata = TableMetadata.minimal("test_ks", "test_table");
+                assertTrue("Controller config file is orphaned after table drop",
+                           Controller.getControllerConfigPath(metadata).exists());
+            });
+
+            // when
+            // stopping the node
+            waitOn(cluster.get(1).shutdown());
+
+            // then
+            // starting the node again should succeed
+            cluster.get(1).startup();
+            // after restart, the orphaned file should be cleaned up
+            cluster.get(1).runOnInstance(() -> {
+                TableMetadata metadata = TableMetadata.minimal("test_ks", "test_table");
+                assertFalse("Controller config file should be deleted after restart cleanup",
+                            Controller.getControllerConfigPath(metadata).exists());
+            });
+        }
+    }
 
     public void vectorControllerConfig(boolean vectorOverride) throws Throwable
     {

test/unit/org/apache/cassandra/db/compaction/CompactionControllerTest.java

@@ -62,6 +62,7 @@
 
 import static org.apache.cassandra.config.CassandraRelevantProperties.ALLOW_UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION;
 import static org.apache.cassandra.db.ColumnFamilyStore.FlushReason.UNIT_TESTS;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotEquals;
@@ -74,12 +75,12 @@ public class CompactionControllerTest extends SchemaLoader
     private static final String KEYSPACE = "CompactionControllerTest";
     private static final String CF1 = "Standard1";
     private static final String CF2 = "Standard2";
-    private static final int TTL_SECONDS = 10;
     private static CountDownLatch compaction2FinishLatch = new CountDownLatch(1);
     private static CountDownLatch createCompactionControllerLatch = new CountDownLatch(1);
     private static CountDownLatch compaction1RefreshLatch = new CountDownLatch(1);
     private static CountDownLatch refreshCheckLatch = new CountDownLatch(1);
     private static int overlapRefreshCounter = 0;
+    private static boolean shouldThrowOnConfigCleanup = false;
 
     @BeforeClass
     public static void defineSchema() throws ConfigurationException
@@ -269,6 +270,34 @@ public void testInFlightBloomFilterMemory()
         }
     }
 
+    @Test
+    @BMRule(name = "Simulate an non standard exception while opening a keyspace",
+    targetClass = "org.apache.cassandra.schema.Schema",
+    targetMethod = "getKeyspaceInstance",
+    targetLocation = "AT ENTRY",
+    condition = "org.apache.cassandra.db.compaction.CompactionControllerTest.shouldThrowOnConfigCleanup",
+    action = "throw new java.lang.RuntimeException(\"Expected exception\");")
+    public void testRethrowingExceptionOnConfigCleanup()
+    {
+        try
+        {
+            // given the config was stored
+            CompactionManager.storeControllerConfig();
+            // below flag is read by byteman, it is needed to throw the exception on the second call of cleanupControllerConfig
+            shouldThrowOnConfigCleanup = true;
+            // when
+            // unhandled exception is throw while cleaning up the controller config
+            // then the excepion is not swallowed
+            assertThatExceptionOfType(RuntimeException.class)
+            .isThrownBy(CompactionManager::cleanupControllerConfig)
+            .withMessage("Expected exception");
+        }
+        finally
+        {
+            shouldThrowOnConfigCleanup = false;
+        }
+    }
+
     @Test
     @BMRules(rules = {
     @BMRule(name = "Pause compaction",