CNDB-13004 don't fail on too long file name for options (#1576)

When a compaction controller config options are being written or read
for a table with long keyspace + table names, FileSystemException
happens.

This commit fixes that such exception is caught and logged. Thus
existing table with long name will continue to function. Tests are 
added.

Also it catches other exceptions.

Author: k-rus

src/java/org/apache/cassandra/db/compaction/unified/AdaptiveController.java

@@ -24,6 +24,9 @@
 import javax.annotation.Nullable;
 
 import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.cassandra.io.FSError;
+import org.apache.cassandra.utils.JVMStabilityInspector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -207,6 +210,15 @@ static Controller fromOptions(Environment env,
         {
             logger.warn("Unable to parse saved options. Using starting value instead:", e);
         }
+        catch (FSError e)
+        {
+            logger.warn("Unable to read controller config file. Using starting value instead:", e);
+        }
+        catch (Throwable e)
+        {
+            logger.warn("Unable to read controller config file. Using starting value instead:", e);
+            JVMStabilityInspector.inspectThrowable(e);
+        }
 
         if (scalingParameters == null)
         {

src/java/org/apache/cassandra/db/compaction/unified/Controller.java

@@ -36,6 +36,8 @@
 import com.google.common.annotations.VisibleForTesting;
 
 import org.apache.cassandra.config.Config;
+import org.apache.cassandra.io.FSError;
+import org.apache.cassandra.utils.JVMStabilityInspector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -423,10 +425,15 @@ public static void storeOptions(String keyspaceName, String tableName, int[] sca
 
             logger.debug(String.format("Writing current scaling parameters and flush size to file %s: %s", f.toPath().toString(), jsonObject));
         }
-        catch(IOException e)
+        catch (IOException | FSError e)
         {
             logger.warn("Unable to save current scaling parameters and flush size. Current controller configuration will be lost if a node restarts: ", e);
         }
+        catch (Throwable e)
+        {
+            logger.warn("Unable to save current scaling parameters and flush size. Current controller configuration will be lost if a node restarts: ", e);
+            JVMStabilityInspector.inspectThrowable(e);
+        }
     }
 
     public abstract void storeControllerConfig();

src/java/org/apache/cassandra/db/compaction/unified/StaticController.java

@@ -25,6 +25,7 @@
 import org.apache.cassandra.db.compaction.CompactionPick;
 import org.apache.cassandra.db.compaction.UnifiedCompactionStrategy;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.io.FSError;
 import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.io.util.FileReader;
 import org.apache.cassandra.utils.JVMStabilityInspector;
@@ -151,6 +152,10 @@ static Controller fromOptions(Environment env,
         {
             logger.warn("Unable to parse saved flush size. Using starting value instead:", e);
         }
+        catch (FSError e)
+        {
+            logger.warn("Unable to read controller config file. Using starting value instead:", e);
+        }
         catch (Throwable e)
         {
             logger.warn("Unable to read controller config file. Using starting value instead:", e);
@@ -235,4 +240,4 @@ public String toString()
                              printScalingParameters(scalingParameters),
                              calculator);
     }
-}
+}

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

@@ -177,6 +177,28 @@ public void testStoreAndCleanupControllerConfig() throws Throwable
         }
     }
 
+    @Test
+    public void testStoreLongTableName() throws Throwable
+    {
+        try (Cluster cluster = init(Cluster.build(1).start()))
+        {
+            cluster.get(1).runOnInstance(() ->
+                                         {
+                                             CompactionManager.storeControllerConfig();
+
+                                             // try to store controller config for a table with a long name
+                                             String keyspaceName = "g38373639353166362d356631322d343864652d393063362d653862616534343165333764_tpch";
+                                             String longTableName = "test_create_k8yq1r75bpzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz";
+                                             int[] scalingParameters = new int[32];
+                                             Arrays.fill(scalingParameters, 5);
+                                             AdaptiveController.storeOptions(keyspaceName, longTableName, scalingParameters, 10 << 20);
+
+                                             // verify that the file wasn't created
+                                             assert !Controller.getControllerConfigPath(keyspaceName, longTableName).exists();
+                                         });
+        }
+    }
+
     @Test
     public void testVectorControllerConfig() throws Throwable
     {
@@ -234,7 +256,6 @@ public void vectorControllerConfig(boolean vectorOverride) throws Throwable
                                                           controller2.getTargetSSTableSize());
                                              assertEquals(0, controller2.getScalingParameter(0));
                                          });
-
         }
     }
 }

test/unit/org/apache/cassandra/db/compaction/unified/AdaptiveControllerTest.java

@@ -97,6 +97,22 @@ private AdaptiveController makeController(long dataSizeGB, int numShards, long s
                                       tableName);
     }
 
+    @Test
+    public void testLongTableNameFromOptions()
+    {
+        String longTableName = "test_create_k8yq1r75bpzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz";
+        when(cfs.getTableName()).thenReturn(longTableName);
+
+        Map<String, String> options = new HashMap<>();
+        options.put(AdaptiveController.THRESHOLD, "0.15");
+        // Calls fromOptions on long table name, which tries to read options from a file.
+        // The too long file name should not lead to a failure.
+        Controller controller = testFromOptions(true, options);
+        assertTrue(controller instanceof AdaptiveController);
+
+        when(cfs.getTableName()).thenReturn(tableName);
+    }
+
     @Test
     public void testFromOptions()
     {