[flink] Refactor CALL procedure "sys.xxx_cluster_config" to support multiple conifg keys

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/GetClusterConfigProcedure.java → fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/GetClusterConfigsProcedure.java

@@ -30,54 +30,61 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 /**
  * Procedure to get cluster configuration(s).
  *
  * <p>This procedure allows querying dynamic cluster configurations. It can retrieve:
  *
  * <ul>
- *   <li>A specific configuration key
+ *   <li>multiple configurations
  *   <li>All configurations (when key parameter is null or empty)
  * </ul>
  *
  * <p>Usage examples:
  *
  * <pre>
  * -- Get a specific configuration
- * CALL sys.get_cluster_config('kv.rocksdb.shared-rate-limiter.bytes-per-sec');
+ * CALL sys.get_cluster_configs('kv.rocksdb.shared-rate-limiter.bytes-per-sec');
+ *
+ * -- Get multiple configurations
+ * CALL sys.get_cluster_configs('kv.rocksdb.shared-rate-limiter.bytes-per-sec', 'datalake.format');
  *
  * -- Get all cluster configurations
- * CALL sys.get_cluster_config();
+ * CALL sys.get_cluster_configs();
  * </pre>
  */
-public class GetClusterConfigProcedure extends ProcedureBase {
+public class GetClusterConfigsProcedure extends ProcedureBase {
 
     @ProcedureHint(
             output =
                     @DataTypeHint(
                             "ROW<config_key STRING, config_value STRING, config_source STRING>"))
     public Row[] call(ProcedureContext context) throws Exception {
-        return getConfigs(null);
+        return getConfigs();
     }
 
     @ProcedureHint(
-            argument = {@ArgumentHint(name = "config_key", type = @DataTypeHint("STRING"))},
+            argument = {@ArgumentHint(name = "config_keys", type = @DataTypeHint("STRING"))},
+            isVarArgs = true,
             output =
                     @DataTypeHint(
                             "ROW<config_key STRING, config_value STRING, config_source STRING>"))
-    public Row[] call(ProcedureContext context, String configKey) throws Exception {
-        return getConfigs(configKey);
+    public Row[] call(ProcedureContext context, String... configKeys) throws Exception {
+        return getConfigs(configKeys);
     }
 
-    private Row[] getConfigs(@Nullable String configKey) throws Exception {
+    private Row[] getConfigs(@Nullable String... configKeys) throws Exception {
         try {
             // Get all cluster configurations
             Collection<ConfigEntry> configs = admin.describeClusterConfigs().get();
 
             List<Row> results = new ArrayList<>();
 
-            if (configKey == null || configKey.isEmpty()) {
+            if (configKeys == null || configKeys.length == 0) {
                 // Return all configurations
                 for (ConfigEntry entry : configs) {
                     results.add(
@@ -87,17 +94,21 @@ private Row[] getConfigs(@Nullable String configKey) throws Exception {
                                     entry.source() != null ? entry.source().name() : "UNKNOWN"));
                 }
             } else {
-                // Find specific configuration
-                for (ConfigEntry entry : configs) {
-                    if (entry.key().equals(configKey)) {
+                // Find configurations
+                // The order of the results is the same as that of the key.
+                Map<String, ConfigEntry> configEntryMap =
+                        configs.stream()
+                                .collect(Collectors.toMap(ConfigEntry::key, Function.identity()));
+                for (String key : configKeys) {
+                    ConfigEntry entry = configEntryMap.get(key);
+                    if (null != entry) {
                         results.add(
                                 Row.of(
                                         entry.key(),
                                         entry.value(),
                                         entry.source() != null
                                                 ? entry.source().name()
                                                 : "UNKNOWN"));
-                        break;
                     }
                 }
             }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ProcedureManager.java

@@ -70,8 +70,9 @@ private enum ProcedureEnum {
         ADD_ACL("sys.add_acl", AddAclProcedure.class),
         DROP_ACL("sys.drop_acl", DropAclProcedure.class),
         List_ACL("sys.list_acl", ListAclProcedure.class),
-        SET_CLUSTER_CONFIG("sys.set_cluster_config", SetClusterConfigProcedure.class),
-        GET_CLUSTER_CONFIG("sys.get_cluster_config", GetClusterConfigProcedure.class),
+        SET_CLUSTER_CONFIGS("sys.set_cluster_configs", SetClusterConfigsProcedure.class),
+        GET_CLUSTER_CONFIGS("sys.get_cluster_configs", GetClusterConfigsProcedure.class),
+        RESET_CLUSTER_CONFIGS("sys.reset_cluster_configs", ResetClusterConfigsProcedure.class),
         ADD_SERVER_TAG("sys.add_server_tag", AddServerTagProcedure.class),
         REMOVE_SERVER_TAG("sys.remove_server_tag", RemoveServerTagProcedure.class);
 

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ResetClusterConfigsProcedure.java

@@ -0,0 +1,106 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.fluss.config.cluster.AlterConfig;
+import org.apache.fluss.config.cluster.AlterConfigOpType;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Procedure to reset cluster configurations to their default values.
+ *
+ * <p>This procedure reverts the configurations to their initial system defaults. The changes are:
+ *
+ * <ul>
+ *   <li>Validated by the CoordinatorServer before persistence
+ *   <li>Persisted in ZooKeeper for durability
+ *   <li>Applied to all relevant servers (Coordinator and TabletServers)
+ *   <li>Survive server restarts
+ * </ul>
+ *
+ * <p>Usage examples:
+ *
+ * <pre>
+ * -- reset a configuration
+ * CALL sys.reset_cluster_configs('kv.rocksdb.shared-rate-limiter.bytes-per-sec');
+ *
+ * -- reset multiple configurations at one time
+ * CALL sys.reset_cluster_configs('kv.rocksdb.shared-rate-limiter.bytes-per-sec', 'datalake.format');
+ *
+ * </pre>
+ *
+ * <p><b>Note:</b> In theory, an operation like <b>Reset to default value</b> should always succeed,
+ * as the default value should be a valid one
+ */
+public class ResetClusterConfigsProcedure extends ProcedureBase {
+
+    @ProcedureHint(
+            argument = {@ArgumentHint(name = "config_keys", type = @DataTypeHint("STRING"))},
+            isVarArgs = true)
+    public String[] call(ProcedureContext context, String... configKeys) throws Exception {
+        try {
+            // Validate config key
+            if (configKeys.length == 0) {
+                throw new IllegalArgumentException(
+                        "config_keys cannot be null or empty. "
+                                + "Please specify valid configuration keys.");
+            }
+
+            List<AlterConfig> configList = new ArrayList<>();
+            List<String> resultMessage = new ArrayList<>();
+
+            for (String key : configKeys) {
+                String configKey = key.trim();
+                if (configKey.isEmpty()) {
+                    throw new IllegalArgumentException(
+                            "Config key cannot be null or empty. "
+                                    + "Please specify a valid configuration key.");
+                }
+
+                String operationDesc = "deleted (reset to default)";
+
+                AlterConfig alterConfig =
+                        new AlterConfig(configKey, null, AlterConfigOpType.DELETE);
+                configList.add(alterConfig);
+                resultMessage.add(
+                        String.format(
+                                "Successfully %s configuration '%s'. ", operationDesc, configKey));
+            }
+
+            // Call Admin API to modify cluster configuration
+            // This will trigger validation on CoordinatorServer before persistence
+            admin.alterClusterConfigs(configList).get();
+
+            return resultMessage.toArray(new String[0]);
+        } catch (IllegalArgumentException e) {
+            // Re-throw validation errors with original message
+            throw e;
+        } catch (Exception e) {
+            // Wrap other exceptions with more context
+            throw new RuntimeException(
+                    String.format("Failed to reset cluster config: %s", e.getMessage()), e);
+        }
+    }
+}