fix comment

Signed-off-by: Pei Yu <[email protected]>

Author: sd4324530

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

@@ -68,13 +68,13 @@ public Row[] call(ProcedureContext context) throws Exception {
     }
 
     @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... configKeys) throws Exception {

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

@@ -29,9 +29,9 @@
 import java.util.List;
 
 /**
- * Procedure to reset cluster configuration dynamically.
+ * Procedure to reset cluster configurations to their default values.
  *
- * <p>This procedure allows modifying dynamic cluster configurations. The changes are:
+ * <p>This procedure reverts the configurations to their initial system defaults. The changes are:
  *
  * <ul>
  *   <li>Validated by the CoordinatorServer before persistence
@@ -51,8 +51,8 @@
  *
  * </pre>
  *
- * <p><b>Note:</b> Not all configurations support dynamic changes. The server will validate the
- * change and reject it if the configuration cannot be reset dynamically.
+ * <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 {
 
@@ -69,7 +69,7 @@ public String[] call(ProcedureContext context, String... configKeys) throws Exce
             }
 
             List<AlterConfig> configList = new ArrayList<>();
-            StringBuilder resultMessage = new StringBuilder();
+            List<String> resultMessage = new ArrayList();
 
             for (String key : configKeys) {
                 String configKey = key.trim();
@@ -84,7 +84,7 @@ public String[] call(ProcedureContext context, String... configKeys) throws Exce
                 AlterConfig alterConfig =
                         new AlterConfig(configKey, null, AlterConfigOpType.DELETE);
                 configList.add(alterConfig);
-                resultMessage.append(
+                resultMessage.add(
                         String.format(
                                 "Successfully %s configuration '%s'. ", operationDesc, configKey));
             }
@@ -93,9 +93,7 @@ public String[] call(ProcedureContext context, String... configKeys) throws Exce
             // This will trigger validation on CoordinatorServer before persistence
             admin.alterClusterConfigs(configList).get();
 
-            return new String[] {
-                resultMessage + "The change is persisted in ZooKeeper and applied to all servers."
-            };
+            return resultMessage.toArray(new String[0]);
         } catch (IllegalArgumentException e) {
             // Re-throw validation errors with original message
             throw e;

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

@@ -49,9 +49,6 @@
  *
  * -- Set multiple configurations at one time
  * CALL sys.set_cluster_configs('kv.rocksdb.shared-rate-limiter.bytes-per-sec', '200MB','datalake.format', 'paimon');
- *
- * -- Delete a configuration (reset to default)
- * CALL sys.set_cluster_configs('kv.rocksdb.shared-rate-limiter.bytes-per-sec', '');
  * </pre>
  *
  * <p><b>Note:</b> Not all configurations support dynamic changes. The server will validate the
@@ -78,7 +75,7 @@ public String[] call(ProcedureContext context, String... configPairs) throws Exc
                                 + "Please specify a valid configuration pairs.");
             }
             List<AlterConfig> configList = new ArrayList<>();
-            StringBuilder resultMessage = new StringBuilder();
+            List<String> resultMessage = new ArrayList<>();
 
             for (int i = 0; i < configPairs.length; i += 2) {
                 String configKey = configPairs[i].trim();
@@ -95,7 +92,7 @@ public String[] call(ProcedureContext context, String... configPairs) throws Exc
                 AlterConfig alterConfig =
                         new AlterConfig(configKey, configValue, AlterConfigOpType.SET);
                 configList.add(alterConfig);
-                resultMessage.append(
+                resultMessage.add(
                         String.format(
                                 "Successfully %s configuration '%s'. ", operationDesc, configKey));
             }
@@ -104,9 +101,7 @@ public String[] call(ProcedureContext context, String... configPairs) throws Exc
             // This will trigger validation on CoordinatorServer before persistence
             admin.alterClusterConfigs(configList).get();
 
-            return new String[] {
-                resultMessage + "The change is persisted in ZooKeeper and applied to all servers."
-            };
+            return resultMessage.toArray(new String[0]);
         } catch (IllegalArgumentException e) {
             // Re-throw validation errors with original message
             throw e;

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java

@@ -284,6 +284,31 @@ void testGetClusterConfigs() throws Exception {
             assertThat(row.getField(2)).isNotNull(); // config_source
         }
 
+        // Get multiple config
+        try (CloseableIterator<Row> resultIterator =
+                tEnv.executeSql(
+                                String.format(
+                                        "Call %s.sys.get_cluster_configs('%s', '%s')",
+                                        CATALOG_NAME,
+                                        ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key(),
+                                        ConfigOptions.DATALAKE_FORMAT.key()))
+                        .collect()) {
+            List<Row> results = CollectionUtil.iteratorToList(resultIterator);
+            assertThat(results).hasSize(2);
+            // the first row
+            Row row0 = results.get(0);
+            assertThat(row0.getField(0))
+                    .isEqualTo(ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key());
+            assertThat(row0.getField(1)).isEqualTo("100 mb");
+            assertThat(row0.getField(2)).isNotNull(); // config_source
+
+            // the second row
+            Row row1 = results.get(1);
+            assertThat(row1.getField(0)).isEqualTo(ConfigOptions.DATALAKE_FORMAT.key());
+            assertThat(row1.getField(1)).isEqualTo("paimon");
+            assertThat(row1.getField(2)).isNotNull(); // config_source
+        }
+
         // Get all configs
         try (CloseableIterator<Row> resultIterator =
                 tEnv.executeSql(String.format("Call %s.sys.get_cluster_configs()", CATALOG_NAME))
@@ -340,11 +365,14 @@ void testSetClusterConfigs() throws Exception {
                                         ConfigOptions.DATALAKE_FORMAT.key()))
                         .collect()) {
             List<Row> results = CollectionUtil.iteratorToList(resultIterator);
-            assertThat(results).hasSize(1);
+            assertThat(results).hasSize(2);
             assertThat(results.get(0).getField(0))
                     .asString()
                     .contains("Successfully set to '300MB'")
-                    .contains(ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key())
+                    .contains(ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key());
+
+            assertThat(results.get(1).getField(0))
+                    .asString()
                     .contains("Successfully set to 'paimon'")
                     .contains(ConfigOptions.DATALAKE_FORMAT.key());
         }
@@ -365,9 +393,10 @@ void testSetClusterConfigs() throws Exception {
         // reset cluster configs.
         tEnv.executeSql(
                         String.format(
-                                "Call %s.sys.reset_cluster_configs('%s')",
+                                "Call %s.sys.reset_cluster_configs('%s', '%s')",
                                 CATALOG_NAME,
-                                ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key()))
+                                ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key(),
+                                ConfigOptions.DATALAKE_FORMAT.key()))
                 .await();
     }
 
@@ -392,11 +421,15 @@ void testResetClusterConfigs() throws Exception {
                                         ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key()))
                         .collect()) {
             List<Row> results = CollectionUtil.iteratorToList(resultIterator);
-            assertThat(results).hasSize(1);
+            assertThat(results).hasSize(2);
             assertThat(results.get(0).getField(0))
                     .asString()
                     .contains("Successfully deleted")
-                    .contains(ConfigOptions.DATALAKE_FORMAT.key())
+                    .contains(ConfigOptions.DATALAKE_FORMAT.key());
+
+            assertThat(results.get(1).getField(0))
+                    .asString()
+                    .contains("Successfully deleted")
                     .contains(ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key());
         }
     }
@@ -429,6 +462,18 @@ void testSetClusterConfigValidation() throws Exception {
                 .rootCause()
                 .hasMessageContaining(
                         "config_pairs must be set in pairs. Please specify a valid configuration pairs");
+
+        // Try to no parameters passed
+        assertThatThrownBy(
+                        () ->
+                                tEnv.executeSql(
+                                                String.format(
+                                                        "Call %s.sys.set_cluster_configs()",
+                                                        CATALOG_NAME))
+                                        .await())
+                .rootCause()
+                .hasMessageContaining(
+                        "config_pairs cannot be null or empty. Please specify a valid configuration pairs");
     }
 
     @Test
@@ -438,12 +483,24 @@ void testResetClusterConfigValidation() throws Exception {
                         () ->
                                 tEnv.executeSql(
                                                 String.format(
-                                                        "Call %s.sys.reset_cluster_configs('')",
+                                                        "Call %s.sys.reset_cluster_configs('invalid.config.key')",
+                                                        CATALOG_NAME))
+                                        .await())
+                .rootCause()
+                .hasMessageContaining(
+                        "The config key invalid.config.key is not allowed to be changed dynamically");
+
+        // Try to no parameters passed
+        assertThatThrownBy(
+                        () ->
+                                tEnv.executeSql(
+                                                String.format(
+                                                        "Call %s.sys.reset_cluster_configs()",
                                                         CATALOG_NAME))
                                         .await())
                 .rootCause()
                 .hasMessageContaining(
-                        "Config key cannot be null or empty. Please specify a valid configuration key");
+                        "config_keys cannot be null or empty. Please specify valid configuration keys");
     }
 
     @ParameterizedTest

website/docs/engine-flink/procedures.md

@@ -162,23 +162,23 @@ CALL sys.list_acl(
 
 Fluss provides procedures to dynamically manage cluster configurations without requiring a server restart.
 
-### get_cluster_config
+### get_cluster_configs
 
 Retrieve cluster configuration values.
 
 **Syntax:**
 
 ```sql
--- Get a specific configuration
-CALL [catalog_name.]sys.get_cluster_config(config_key => 'STRING')
+-- Get multiple configurations
+CALL [catalog_name.]sys.get_cluster_configs(config_keys => 'key1' [, 'key2', ...])
 
 -- Get all cluster configurations
-CALL [catalog_name.]sys.get_cluster_config()
+CALL [catalog_name.]sys.get_cluster_configs()
 ```
 
 **Parameters:**
 
-- `config_key` (optional): The configuration key to retrieve. If omitted, returns all cluster configurations.
+- `config_keys` (optional): The configuration keys to retrieve. If omitted, returns all cluster configurations.
 
 **Returns:** A table with columns:
 - `config_key`: The configuration key name
@@ -192,35 +192,35 @@ CALL [catalog_name.]sys.get_cluster_config()
 USE fluss_catalog;
 
 -- Get a specific configuration
-CALL sys.get_cluster_config(
-  config_key => 'kv.rocksdb.shared-rate-limiter.bytes-per-sec'
+CALL sys.get_cluster_configs(
+  config_keys => 'kv.rocksdb.shared-rate-limiter.bytes-per-sec'
+);
+
+-- Get multiple configuration
+CALL sys.get_cluster_configs(
+  config_keys => 'kv.rocksdb.shared-rate-limiter.bytes-per-sec', 'datalake.format'
 );
 
 -- Get all cluster configurations
-CALL sys.get_cluster_config();
+CALL sys.get_cluster_configs();
 ```
 
-### set_cluster_config
+### set_cluster_configs
 
-Set or delete a cluster configuration dynamically.
+Set cluster configurations dynamically.
 
 **Syntax:**
 
 ```sql
--- Set a configuration value
-CALL [catalog_name.]sys.set_cluster_config(
-  config_key => 'STRING',
-  config_value => 'STRING'
+-- Set configuration values
+CALL [catalog_name.]sys.set_cluster_configs(
+  config_pairs => 'key1', 'value1' [, 'key2', 'value2' ...]
 )
-
--- Delete a configuration (reset to default)
-CALL [catalog_name.]sys.set_cluster_config(config_key => 'STRING')
 ```
 
 **Parameters:**
 
-- `config_key` (required): The configuration key to modify.
-- `config_value` (optional): The new value to set. If omitted or empty, the configuration is deleted (reset to default).
+- `config_pairs`(required): For key-value pairs in configuration items, the number of parameters must be even.
 
 **Important Notes:**
 
@@ -236,20 +236,51 @@ CALL [catalog_name.]sys.set_cluster_config(config_key => 'STRING')
 USE fluss_catalog;
 
 -- Set RocksDB rate limiter
-CALL sys.set_cluster_config(
-  config_key => 'kv.rocksdb.shared-rate-limiter.bytes-per-sec',
-  config_value => '200MB'
+CALL sys.set_cluster_configs(
+  config_pairs => 'kv.rocksdb.shared-rate-limiter.bytes-per-sec', '200MB'
 );
 
--- Set datalake format
-CALL sys.set_cluster_config(
-  config_key => 'datalake.format',
-  config_value => 'paimon'
+-- Set RocksDB rate limiter and datalake format
+CALL sys.set_cluster_configs(
+  config_pairs => 'kv.rocksdb.shared-rate-limiter.bytes-per-sec', '200MB', 'datalake.format','paimon'
 );
+```
 
--- Delete a configuration (reset to default)
-CALL sys.set_cluster_config(
-  config_key => 'kv.rocksdb.shared-rate-limiter.bytes-per-sec'
-);
+### reset_cluster_configs
+
+reset cluster configurations dynamically.
+
+**Syntax:**
+
+```sql
+-- reset configuration values
+CALL [catalog_name.]sys.reset_cluster_configs(config_keys => 'key1' [, 'key2', ...])
 ```
 
+**Parameters:**
+
+- `config_keys`(required): The configuration keys to reset.
+
+**Important Notes:**
+
+- Changes are validated before being applied and persisted in ZooKeeper
+- Changes are automatically applied to all servers (Coordinator and TabletServers)
+- Changes survive server restarts
+- Not all configurations support dynamic changes. The server will reject invalid modifications
+
+**Example:**
+
+```sql title="Flink SQL"
+-- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
+USE fluss_catalog;
+
+-- Reset a specific configuration
+CALL sys.reset_cluster_configs(
+  config_keys => 'kv.rocksdb.shared-rate-limiter.bytes-per-sec'
+);
+
+-- Reset RocksDB rate limiter and datalake format
+CALL sys.reset_cluster_configs(
+  config_keys => 'kv.rocksdb.shared-rate-limiter.bytes-per-sec', 'datalake.format'
+);
+```