support generic flink procedure

Author: platinumhamburg

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/GetSharedRocksDBRateLimiterProcedure.java

@@ -70,10 +70,8 @@ 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_SHARED_ROCKSDB_RATE_LIMITER(
-                "sys.set_shared_rocksdb_rate_limiter", SetSharedRocksDBRateLimiterProcedure.class),
-        GET_SHARED_ROCKSDB_RATE_LIMITER(
-                "sys.get_shared_rocksdb_rate_limiter", GetSharedRocksDBRateLimiterProcedure.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;