Support shared RocksDB rate limiter in Fluss

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

@@ -1524,6 +1524,16 @@ public class ConfigOptions {
                             "The max size of the consumed memory for RocksDB batch write, "
                                     + "will flush just based on item count if this config set to 0.");
 
+    public static final ConfigOption<MemorySize> KV_SHARED_RATE_LIMITER_BYTES_PER_SEC =
+            key("kv.rocksdb.shared-rate-limiter.bytes-per-sec")
+                    .memoryType()
+                    .defaultValue(MemorySize.parse("100mb"))
+                    .withDescription(
+                            "The shared rate limit in bytes per second for RocksDB flush and compaction operations "
+                                    + "across all RocksDB instances in the TabletServer. "
+                                    + "All KV tablets share a single global RateLimiter to prevent disk IO from being saturated. "
+                                    + "The default value is `100MB/s`. Set to 0 to disable rate limiting.");
+
     // --------------------------------------------------------------------------
     // Provided configurable ColumnFamilyOptions within Fluss
     // --------------------------------------------------------------------------

fluss-common/src/main/java/org/apache/fluss/config/cluster/ConfigValidator.java

@@ -0,0 +1,72 @@
+/*
+ * 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.config.cluster;
+
+import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.exception.ConfigException;
+
+import javax.annotation.Nullable;
+
+/**
+ * Validator for a single dynamic configuration key.
+ *
+ * <p>Unlike {@link ServerReconfigurable}, validators are stateless and only perform validation
+ * logic without requiring component instances. This allows coordinators to validate configurations
+ * for components they don't run (e.g., KvManager).
+ *
+ * <p>Example use case: CoordinatorServer needs to validate KV-related configurations even though it
+ * doesn't have a KvManager instance. A {@link ConfigValidator} can be registered on both
+ * CoordinatorServer (for validation) and TabletServer (for both validation and actual
+ * reconfiguration via {@link ServerReconfigurable}).
+ *
+ * <p>Each validator monitors a single configuration key. The validator will only be invoked when
+ * that specific key changes, improving validation efficiency.
+ *
+ * <p>This interface is designed to be stateless and thread-safe. Implementations should not rely on
+ * any mutable component state.
+ */
+@PublicEvolving
+public interface ConfigValidator {
+
+    /**
+     * Returns the configuration key this validator monitors.
+     *
+     * <p>The validator will only be invoked when this specific configuration key changes. This
+     * allows efficient filtering of validators and clear declaration of dependencies.
+     *
+     * @return the configuration key to monitor, must not be null or empty
+     */
+    String configKey();
+
+    /**
+     * Validates a configuration value change.
+     *
+     * <p>This method is called when the monitored configuration key changes. It should check
+     * whether the new value is valid, potentially considering the old value and validation rules.
+     *
+     * <p>The method should be stateless and deterministic - given the same old and new values, it
+     * should always produce the same validation result.
+     *
+     * @param oldValue the previous value of the configuration key, null if the key was not set
+     *     before
+     * @param newValue the new value of the configuration key, null if the key is being deleted
+     * @throws ConfigException if the configuration change is invalid, with a descriptive error
+     *     message explaining why the change cannot be applied
+     */
+    void validate(@Nullable String oldValue, @Nullable String newValue) throws ConfigException;
+}

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

@@ -0,0 +1,124 @@
+/*
+ * 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.ConfigEntry;
+
+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 org.apache.flink.types.Row;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * Procedure to get cluster configuration(s).
+ *
+ * <p>This procedure allows querying dynamic cluster configurations. It can retrieve:
+ *
+ * <ul>
+ *   <li>A specific configuration key
+ *   <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.shared-rate-limiter.bytes-per-sec');
+ *
+ * -- Get all cluster configurations
+ * CALL sys.get_cluster_config();
+ * </pre>
+ */
+public class GetClusterConfigProcedure 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);
+    }
+
+    @ProcedureHint(
+            argument = {@ArgumentHint(name = "config_key", type = @DataTypeHint("STRING"))},
+            output =
+                    @DataTypeHint(
+                            "ROW<config_key STRING, config_value STRING, config_source STRING>"))
+    public Row[] call(ProcedureContext context, String configKey) throws Exception {
+        return getConfigs(configKey);
+    }
+
+    private Row[] getConfigs(@Nullable String configKey) throws Exception {
+        try {
+            // Get all cluster configurations
+            Collection<ConfigEntry> configs = admin.describeClusterConfigs().get();
+
+            List<Row> results = new ArrayList<>();
+
+            if (configKey == null || configKey.isEmpty()) {
+                // Return all configurations
+                for (ConfigEntry entry : configs) {
+                    results.add(
+                            Row.of(
+                                    entry.key(),
+                                    entry.value(),
+                                    entry.source() != null ? entry.source().name() : "UNKNOWN"));
+                }
+
+                if (results.isEmpty()) {
+                    return new Row[] {Row.of("No cluster configurations found", null, null)};
+                }
+            } else {
+                // Find specific configuration
+                for (ConfigEntry entry : configs) {
+                    if (entry.key().equals(configKey)) {
+                        results.add(
+                                Row.of(
+                                        entry.key(),
+                                        entry.value(),
+                                        entry.source() != null
+                                                ? entry.source().name()
+                                                : "UNKNOWN"));
+                        break;
+                    }
+                }
+
+                if (results.isEmpty()) {
+                    return new Row[] {
+                        Row.of(
+                                String.format("Configuration key '%s' not found", configKey),
+                                null,
+                                null)
+                    };
+                }
+            }
+
+            return results.toArray(new Row[0]);
+
+        } catch (Exception e) {
+            throw new RuntimeException("Failed to get cluster config: " + e.getMessage(), e);
+        }
+    }
+}

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

@@ -69,7 +69,9 @@ private static Map<String, Class<? extends ProcedureBase>> initProcedureMap() {
     private enum ProcedureEnum {
         ADD_ACL("sys.add_acl", AddAclProcedure.class),
         DROP_ACL("sys.drop_acl", DropAclProcedure.class),
-        List_ACL("sys.list_acl", ListAclProcedure.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);
 
         private final String path;
         private final Class<? extends ProcedureBase> procedureClass;

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

@@ -0,0 +1,128 @@
+/*
+ * 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 javax.annotation.Nullable;
+
+import java.util.Collections;
+
+/**
+ * Procedure to set or delete cluster configuration dynamically.
+ *
+ * <p>This procedure allows modifying dynamic cluster configurations. 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>
+ * -- Set a configuration
+ * CALL sys.set_cluster_config('kv.shared-rate-limiter.bytes-per-sec', '200MB');
+ * CALL sys.set_cluster_config('datalake.format', 'paimon');
+ *
+ * -- Delete a configuration (reset to default)
+ * CALL sys.set_cluster_config('kv.shared-rate-limiter.bytes-per-sec', NULL);
+ * CALL sys.set_cluster_config('kv.shared-rate-limiter.bytes-per-sec', '');
+ * </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 modified dynamically or if the new value is
+ * invalid.
+ */
+public class SetClusterConfigProcedure extends ProcedureBase {
+
+    @ProcedureHint(argument = {@ArgumentHint(name = "config_key", type = @DataTypeHint("STRING"))})
+    public String[] call(ProcedureContext context, String configKey) throws Exception {
+        return performSet(configKey, null);
+    }
+
+    @ProcedureHint(
+            argument = {
+                @ArgumentHint(name = "config_key", type = @DataTypeHint("STRING")),
+                @ArgumentHint(name = "config_value", type = @DataTypeHint("STRING"))
+            })
+    public String[] call(ProcedureContext context, String configKey, String configValue)
+            throws Exception {
+        return performSet(configKey, configValue);
+    }
+
+    private String[] performSet(String configKey, @Nullable String configValue) throws Exception {
+
+        try {
+            // Validate config key
+            if (configKey == null || configKey.trim().isEmpty()) {
+                throw new IllegalArgumentException(
+                        "Config key cannot be null or empty. "
+                                + "Please specify a valid configuration key.");
+            }
+
+            configKey = configKey.trim();
+
+            // Determine operation type
+            AlterConfigOpType opType;
+            String operationDesc;
+
+            if (configValue == null || configValue.trim().isEmpty()) {
+                // Delete operation - reset to default
+                opType = AlterConfigOpType.DELETE;
+                operationDesc = "deleted (reset to default)";
+            } else {
+                // Set operation
+                opType = AlterConfigOpType.SET;
+                operationDesc = String.format("set to '%s'", configValue);
+            }
+
+            // Construct configuration modification operation.
+            AlterConfig alterConfig = new AlterConfig(configKey, configValue, opType);
+
+            // Call Admin API to modify cluster configuration
+            // This will trigger validation on CoordinatorServer before persistence
+            admin.alterClusterConfigs(Collections.singletonList(alterConfig)).get();
+
+            return new String[] {
+                String.format(
+                        "Successfully %s configuration '%s'. "
+                                + "The change is persisted in ZooKeeper and applied to all servers.",
+                        operationDesc, configKey)
+            };
+
+        } 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 set cluster config '%s': %s", configKey, e.getMessage()),
+                    e);
+        }
+    }
+}