Support shared RocksDB rate limiter in Fluss

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

@@ -28,10 +28,12 @@
 import org.apache.fluss.metadata.MergeEngineType;
 import org.apache.fluss.utils.ArrayUtils;
 
+import java.lang.reflect.Field;
 import java.time.Duration;
 import java.time.ZoneId;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -1576,6 +1578,17 @@ 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(new MemorySize(Long.MAX_VALUE))
+                    .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 RateLimiter is always enabled. The default value is Long.MAX_VALUE (effectively unlimited). "
+                                    + "Set to a lower value (e.g., 100MB) to limit the rate.");
+
     // --------------------------------------------------------------------------
     // Provided configurable ColumnFamilyOptions within Fluss
     // --------------------------------------------------------------------------
@@ -1869,4 +1882,45 @@ public enum KvCompressionType {
         LZ4,
         ZSTD
     }
+
+    // ------------------------------------------------------------------------
+    //  ConfigOptions Registry and Utilities
+    // ------------------------------------------------------------------------
+
+    /**
+     * Holder class for lazy initialization of ConfigOptions registry. This ensures that the
+     * registry is initialized only when first accessed, and guarantees that all ConfigOption fields
+     * are already initialized (since static initialization happens in declaration order).
+     */
+    private static class ConfigOptionsHolder {
+        private static final Map<String, ConfigOption<?>> CONFIG_OPTIONS_BY_KEY;
+
+        static {
+            Map<String, ConfigOption<?>> options = new HashMap<>();
+            Field[] fields = ConfigOptions.class.getFields();
+            for (Field field : fields) {
+                if (!ConfigOption.class.isAssignableFrom(field.getType())) {
+                    continue;
+                }
+                try {
+                    ConfigOption<?> configOption = (ConfigOption<?>) field.get(null);
+                    options.put(configOption.key(), configOption);
+                } catch (IllegalAccessException e) {
+                    // Ignore fields that cannot be accessed
+                }
+            }
+            CONFIG_OPTIONS_BY_KEY = Collections.unmodifiableMap(options);
+        }
+    }
+
+    /**
+     * Gets the ConfigOption for a given key.
+     *
+     * @param key the configuration key
+     * @return the ConfigOption if found, null otherwise
+     */
+    @Internal
+    public static ConfigOption<?> getConfigOption(String key) {
+        return ConfigOptionsHolder.CONFIG_OPTIONS_BY_KEY.get(key);
+    }
 }

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

@@ -0,0 +1,78 @@
+/*
+ * 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.
+ *
+ * <p><b>Type-safe validation:</b> The validator receives strongly-typed values that have already
+ * been parsed and validated for basic type correctness. This avoids redundant string parsing and
+ * allows validators to focus on business logic validation.
+ *
+ * @param <T> the type of the configuration value being validated
+ */
+@PublicEvolving
+public interface ConfigValidator<T> {
+
+    /**
+     * 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 T oldValue, @Nullable T newValue) throws ConfigException;
+}

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

@@ -0,0 +1,111 @@
+/*
+ * 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.rocksdb.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"));
+                }
+            } 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;
+                    }
+                }
+            }
+
+            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;