[server] Introduce plugin.classloader.parent-first-patterns.additional configuration (#1222)

Author: luoyuxia

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -112,7 +112,21 @@ public class ConfigOptions {
                     .withDescription(
                             "A (semicolon-separated) list of patterns that specifies which classes should always be"
                                     + " resolved through the plugin parent ClassLoader first. A pattern is a simple prefix that is checked "
-                                    + " against the fully qualified class name. This setting should generally not be modified.");
+                                    + " against the fully qualified class name. This setting should generally not be modified. To add another "
+                                    + " pattern we recommend to use \"plugin.classloader.parent-first-patterns.additional\" instead.");
+
+    public static final ConfigOption<List<String>>
+            PLUGIN_ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL =
+                    key("plugin.classloader.parent-first-patterns.additional")
+                            .stringType()
+                            .asList()
+                            .defaultValues()
+                            .withDescription(
+                                    "A (semicolon-separated) list of patterns that specifies which classes should always be"
+                                            + " resolved through the plugin parent ClassLoader first. A pattern is a simple prefix that is checked "
+                                            + " against the fully qualified class name. These patterns are appended to \""
+                                            + PLUGIN_ALWAYS_PARENT_FIRST_LOADER_PATTERNS.key()
+                                            + "\".");
 
     public static final ConfigOption<Duration> AUTO_PARTITION_CHECK_INTERVAL =
             key("auto-partition.check.interval")

fluss-common/src/main/java/org/apache/fluss/plugin/PluginConfig.java

@@ -17,16 +17,20 @@
 
 package org.apache.fluss.plugin;
 
-import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.shaded.guava32.com.google.common.collect.Iterables;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.nio.file.Path;
+import java.util.List;
 import java.util.Optional;
 
+import static org.apache.fluss.config.ConfigOptions.PLUGIN_ALWAYS_PARENT_FIRST_LOADER_PATTERNS;
+import static org.apache.fluss.config.ConfigOptions.PLUGIN_ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL;
+
 /** Stores the configuration for plugins mechanism. */
 @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
 public class PluginConfig {
@@ -62,9 +66,13 @@ public String[] getAlwaysParentFirstPatterns() {
     public static PluginConfig fromConfiguration(Configuration configuration) {
         return new PluginConfig(
                 getPluginsDir().map(File::toPath),
-                configuration
-                        .get(ConfigOptions.PLUGIN_ALWAYS_PARENT_FIRST_LOADER_PATTERNS)
-                        .toArray(new String[0]));
+                getPluginParentFirstLoaderPatterns(configuration));
+    }
+
+    private static String[] getPluginParentFirstLoaderPatterns(Configuration config) {
+        List<String> base = config.get(PLUGIN_ALWAYS_PARENT_FIRST_LOADER_PATTERNS);
+        List<String> append = config.get(PLUGIN_ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL);
+        return Iterables.toArray(Iterables.concat(base, append), String.class);
     }
 
     public static Optional<File> getPluginsDir() {

fluss-common/src/test/java/org/apache/fluss/plugin/PluginConfigTest.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.plugin;
+
+import org.apache.fluss.config.Configuration;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.apache.fluss.config.ConfigOptions.PLUGIN_ALWAYS_PARENT_FIRST_LOADER_PATTERNS;
+import static org.apache.fluss.config.ConfigOptions.PLUGIN_ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link PluginConfig} . */
+class PluginConfigTest {
+
+    @Test
+    void testFromConfig() {
+        // only set parent-first-patterns.default
+        Configuration config = new Configuration();
+        config.set(
+                PLUGIN_ALWAYS_PARENT_FIRST_LOADER_PATTERNS,
+                Arrays.asList("org.apache.fluss.", "java."));
+        PluginConfig pluginConfig = PluginConfig.fromConfiguration(config);
+        assertThat(pluginConfig.getAlwaysParentFirstPatterns())
+                .isEqualTo(new String[] {"org.apache.fluss.", "java."});
+        // also set parent-first-patterns.additional
+        config.set(
+                PLUGIN_ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL,
+                Collections.singletonList("org.apache.hadoop."));
+        pluginConfig = PluginConfig.fromConfiguration(config);
+        assertThat(pluginConfig.getAlwaysParentFirstPatterns())
+                .isEqualTo(new String[] {"org.apache.fluss.", "java.", "org.apache.hadoop."});
+    }
+}