You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "platinumhamburg (via GitHub)" <gi...@apache.org> on 2025/12/15 08:48:23 UTC
[PR] Support shared RocksDB rate limiter in Fluss [fluss]
platinumhamburg opened a new pull request, #2178:
URL: https://github.com/apache/fluss/pull/2178
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
**(The sections below can be removed for hotfixes or typos)**
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #2177
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support shared RocksDB rate limiter in Fluss [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on PR #2178:
URL: https://github.com/apache/fluss/pull/2178#issuecomment-3686725218
> Thanks @platinumhamburg , this PR is already in a good shape. I only left some minor comments.
Thanks for your detailed review, all comments resolved now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support shared RocksDB rate limiter in Fluss [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on code in PR #2178:
URL: https://github.com/apache/fluss/pull/2178#discussion_r2629318433
##########
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', '');
Review Comment:
Maybe `set_cluster_config` and `delete_cluster_config` can be divided into different procedures?
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java:
##########
@@ -86,12 +87,42 @@ void before() throws ExecutionException, InterruptedException {
tEnv.executeSql("use catalog " + CATALOG_NAME);
}
+ @AfterEach
+ void after() throws Exception {
+ // Clean up any dynamic config changes made during the test
+ // to ensure tests don't affect each other
+ if (tEnv != null) {
+ try {
+ // Delete dynamic configs that might have been modified during tests
+ // This resets them to their initial static configuration values
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.set_cluster_config('%s')",
+ CATALOG_NAME,
+ ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key()))
+ .await();
+ tEnv.executeSql(
+ String.format(
+ "Call %s.sys.set_cluster_config('%s')",
+ CATALOG_NAME, ConfigOptions.DATALAKE_FORMAT.key()))
+ .await();
+ } catch (Exception e) {
+ // Ignore cleanup errors to avoid masking test failures
+ }
+ }
Review Comment:
Can these be moved into test cases related to set cluster configuration? It feels a bit odd to place them in afterEach？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support shared RocksDB rate limiter in Fluss [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #2178:
URL: https://github.com/apache/fluss/pull/2178#discussion_r2621702566
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/SetSharedRocksDBRateLimiterProcedure.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.ConfigOptions;
+import org.apache.fluss.config.MemorySize;
+import org.apache.fluss.config.cluster.AlterConfig;
+import org.apache.fluss.config.cluster.AlterConfigOpType;
+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 java.util.Collection;
+import java.util.Collections;
+
+/**
+ * Procedure to set RocksDB rate limiter dynamically via cluster configuration. The configuration
+ * will be persisted in ZooKeeper and applied to all TabletServers.
+ *
+ * <p>Usage examples:
+ *
+ * <pre>
+ * -- Set rate limiter to 200MB/s
+ * CALL sys.set_shared_rocksdb_rate_limiter('200MB');
+ *
+ * -- Set rate limiter to 1GB/s
+ * CALL sys.set_shared_rocksdb_rate_limiter('1GB');
+ * </pre>
+ */
+public class SetSharedRocksDBRateLimiterProcedure extends ProcedureBase {
+
+ @ProcedureHint(
+ argument = {@ArgumentHint(name = "bytes_per_second", type = @DataTypeHint("STRING"))})
+ public String[] call(ProcedureContext context, String bytesPerSecondStr) throws Exception {
+
+ try {
+ // Parse and validate input format first
+ long bytesPerSecond = MemorySize.parse(bytesPerSecondStr).getBytes();
+
+ // Get cluster configuration
+ Collection<ConfigEntry> configs = admin.describeClusterConfigs().get();
+
+ ConfigEntry configEntry = null;
+ for (ConfigEntry entry : configs) {
+ if (entry.key().equals(ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key())) {
+ configEntry = entry;
+ }
+ }
+
+ if (null == configEntry) {
+ throw new IllegalStateException(
+ "Failed to set shared RocksDB rate limiter: previous config not found, server does not support rate limiter setting.");
+ }
+
+ long oldValue = MemorySize.parseBytes(configEntry.value());
Review Comment:
This can move to server, client just need to alter config. All the clients can reuse it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support shared RocksDB rate limiter in Fluss [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on code in PR #2178:
URL: https://github.com/apache/fluss/pull/2178#discussion_r2642918423
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBResourceContainer.java:
##########
@@ -73,29 +74,41 @@ public class RocksDBResourceContainer implements AutoCloseable {
private final boolean enableStatistics;
+ /** The shared rate limiter for all RocksDB instances. */
+ private final RateLimiter sharedRateLimiter;
+
/** The handles to be closed when the container is closed. */
private final ArrayList<AutoCloseable> handlesToClose;
@VisibleForTesting
RocksDBResourceContainer() {
- this(new Configuration(), null, false);
+ this(new Configuration(), null, false, null);
}
public RocksDBResourceContainer(ReadableConfig configuration, @Nullable File instanceBasePath) {
- this(configuration, instanceBasePath, false);
+ this(configuration, instanceBasePath, false, null);
}
public RocksDBResourceContainer(
ReadableConfig configuration,
@Nullable File instanceBasePath,
boolean enableStatistics) {
+ this(configuration, instanceBasePath, enableStatistics, null);
Review Comment:
Since RocksDBResourceContainer is instance-scoped, a non-shared RateLimiter is meaningless and cannot effectively control service-level traffic. I will remove the overloaded constructor that doesn’t accept a RateLimiter, and instead create a shared RateLimiter at the KvManager layer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support shared RocksDB rate limiter in Fluss [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on code in PR #2178:
URL: https://github.com/apache/fluss/pull/2178#discussion_r2622861962
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/SetSharedRocksDBRateLimiterProcedure.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.ConfigOptions;
+import org.apache.fluss.config.MemorySize;
+import org.apache.fluss.config.cluster.AlterConfig;
+import org.apache.fluss.config.cluster.AlterConfigOpType;
+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 java.util.Collection;
+import java.util.Collections;
+
+/**
+ * Procedure to set RocksDB rate limiter dynamically via cluster configuration. The configuration
+ * will be persisted in ZooKeeper and applied to all TabletServers.
+ *
+ * <p>Usage examples:
+ *
+ * <pre>
+ * -- Set rate limiter to 200MB/s
+ * CALL sys.set_shared_rocksdb_rate_limiter('200MB');
+ *
+ * -- Set rate limiter to 1GB/s
+ * CALL sys.set_shared_rocksdb_rate_limiter('1GB');
+ * </pre>
+ */
+public class SetSharedRocksDBRateLimiterProcedure extends ProcedureBase {
Review Comment:
I reimplemented a generic one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support shared RocksDB rate limiter in Fluss [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on code in PR #2178:
URL: https://github.com/apache/fluss/pull/2178#discussion_r2638403534
##########
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)};
Review Comment:
I'll return empty rows directly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support shared RocksDB rate limiter in Fluss [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on PR #2178:
URL: https://github.com/apache/fluss/pull/2178#issuecomment-3684907544
I've updated this PR and resolved all comments above, thanks for review again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support shared RocksDB rate limiter in Fluss [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on code in PR #2178:
URL: https://github.com/apache/fluss/pull/2178#discussion_r2621885125
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/GetSharedRocksDBRateLimiterProcedure.java:
##########
@@ -0,0 +1,86 @@
+/*
+ * 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.ConfigOptions;
+import org.apache.fluss.config.MemorySize;
+import org.apache.fluss.config.cluster.ConfigEntry;
+
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import java.util.Collection;
+
+/**
+ * Procedure to get current RocksDB rate limiter configuration from cluster config.
+ *
+ * <p>Usage:
+ *
+ * <pre>
+ * CALL sys.get_shared_rocksdb_rate_limiter();
+ * </pre>
+ */
+public class GetSharedRocksDBRateLimiterProcedure extends ProcedureBase {
Review Comment:
Why need this? In my side, this parameter is only useful to cluster administrators. If it's really necessary to introduce it, I personally think it should also be abstracted into a "describe config" (i.e., show all configs) operation.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/SetSharedRocksDBRateLimiterProcedure.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.ConfigOptions;
+import org.apache.fluss.config.MemorySize;
+import org.apache.fluss.config.cluster.AlterConfig;
+import org.apache.fluss.config.cluster.AlterConfigOpType;
+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 java.util.Collection;
+import java.util.Collections;
+
+/**
+ * Procedure to set RocksDB rate limiter dynamically via cluster configuration. The configuration
+ * will be persisted in ZooKeeper and applied to all TabletServers.
+ *
+ * <p>Usage examples:
+ *
+ * <pre>
+ * -- Set rate limiter to 200MB/s
+ * CALL sys.set_shared_rocksdb_rate_limiter('200MB');
+ *
+ * -- Set rate limiter to 1GB/s
+ * CALL sys.set_shared_rocksdb_rate_limiter('1GB');
+ * </pre>
+ */
+public class SetSharedRocksDBRateLimiterProcedure extends ProcedureBase {
Review Comment:
It can be abstracted into an `AlterConfigProcedure`, which can then be reused for modifying other cluster parameter.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support shared RocksDB rate limiter in Fluss [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on code in PR #2178:
URL: https://github.com/apache/fluss/pull/2178#discussion_r2622860727
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/SetSharedRocksDBRateLimiterProcedure.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.ConfigOptions;
+import org.apache.fluss.config.MemorySize;
+import org.apache.fluss.config.cluster.AlterConfig;
+import org.apache.fluss.config.cluster.AlterConfigOpType;
+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 java.util.Collection;
+import java.util.Collections;
+
+/**
+ * Procedure to set RocksDB rate limiter dynamically via cluster configuration. The configuration
+ * will be persisted in ZooKeeper and applied to all TabletServers.
+ *
+ * <p>Usage examples:
+ *
+ * <pre>
+ * -- Set rate limiter to 200MB/s
+ * CALL sys.set_shared_rocksdb_rate_limiter('200MB');
+ *
+ * -- Set rate limiter to 1GB/s
+ * CALL sys.set_shared_rocksdb_rate_limiter('1GB');
+ * </pre>
+ */
+public class SetSharedRocksDBRateLimiterProcedure extends ProcedureBase {
+
+ @ProcedureHint(
+ argument = {@ArgumentHint(name = "bytes_per_second", type = @DataTypeHint("STRING"))})
+ public String[] call(ProcedureContext context, String bytesPerSecondStr) throws Exception {
+
+ try {
+ // Parse and validate input format first
+ long bytesPerSecond = MemorySize.parse(bytesPerSecondStr).getBytes();
+
+ // Get cluster configuration
+ Collection<ConfigEntry> configs = admin.describeClusterConfigs().get();
+
+ ConfigEntry configEntry = null;
+ for (ConfigEntry entry : configs) {
+ if (entry.key().equals(ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key())) {
+ configEntry = entry;
+ }
+ }
+
+ if (null == configEntry) {
+ throw new IllegalStateException(
+ "Failed to set shared RocksDB rate limiter: previous config not found, server does not support rate limiter setting.");
+ }
+
+ long oldValue = MemorySize.parseBytes(configEntry.value());
Review Comment:
Thanks for comment, I reimplemented a generic procedure, and moved this validation to server side.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support shared RocksDB rate limiter in Fluss [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #2178:
URL: https://github.com/apache/fluss/pull/2178#discussion_r2642532582
##########
fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java:
##########
@@ -191,7 +191,8 @@ private KvTablet createKvTablet(
rowMerger,
DEFAULT_COMPRESSION,
schemaGetter,
- tableConf.getChangelogImage());
+ tableConf.getChangelogImage(),
+ null);
Review Comment:
ditto
##########
website/docs/engine-flink/ddl.md:
##########
@@ -406,3 +406,252 @@ WITH (
- Only continuous refresh mode is supported
- Schema is automatically derived from the query
- Materialized tables are stored as regular Fluss tables with special metadata
+
+## Procedures
Review Comment:
What I mean is to place the "Procedures" documentation as a separate page under the "DDL" page. Additionally, each procedure method should be formatted as an H3 or H2 header, so it appears in the TOC (table of contents) and can be directly linked via URL.
##########
fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java:
##########
@@ -115,68 +148,296 @@ boolean isAllowedConfig(String key) {
private void updateCurrentConfig(Map<String, String> newDynamicConfigs, boolean skipErrorConfig)
throws Exception {
- Map<String, String> newProps = new HashMap<>(initialConfigMap);
- overrideProps(newProps, newDynamicConfigs);
- Configuration newConfig = Configuration.fromMap(newProps);
- Configuration oldConfig = currentConfig;
- Set<ServerReconfigurable> appliedServerReconfigurableSet = new HashSet<>();
- if (!newProps.equals(currentConfigMap)) {
- serverReconfigures
- .values()
- .forEach(
- serverReconfigurable -> {
- try {
- serverReconfigurable.validate(newConfig);
- } catch (ConfigException e) {
- LOG.error(
- "Validate new dynamic config error and will roll back all the applied config.",
- e);
- if (!skipErrorConfig) {
- throw e;
- }
- }
- });
-
- Exception throwable = null;
- for (ServerReconfigurable serverReconfigurable : serverReconfigures.values()) {
- try {
- serverReconfigurable.reconfigure(newConfig);
- appliedServerReconfigurableSet.add(serverReconfigurable);
- } catch (ConfigException e) {
- LOG.error(
- "Apply new dynamic error and will roll back all the applied config.",
- e);
- if (!skipErrorConfig) {
- throwable = e;
- break;
- }
- }
+ // Compute effective config changes (merge with initial configs)
+ Map<String, String> effectiveChanges =
+ computeEffectiveChanges(newDynamicConfigs, skipErrorConfig);
+
+ // Early return if no effective changes
+ if (effectiveChanges.isEmpty()) {
+ return;
+ }
+
+ // Build new configuration by merging initial + dynamic configs
+ Map<String, String> newConfigMap = buildConfigMap(effectiveChanges);
+ Configuration newConfig = Configuration.fromMap(newConfigMap);
+
+ // Apply changes to all registered ServerReconfigurable instances
+ applyToServerReconfigurables(newConfig, skipErrorConfig);
+
+ // Update internal state
+ updateInternalState(newConfig, newConfigMap, newDynamicConfigs);
+ LOG.info("Dynamic configs changed: {}", newDynamicConfigs);
Review Comment:
I think logging the the chagned configs `effectiveChanges` is more helpful than the full set `newDynamicConfigs`.
##########
fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java:
##########
@@ -115,68 +148,296 @@ boolean isAllowedConfig(String key) {
private void updateCurrentConfig(Map<String, String> newDynamicConfigs, boolean skipErrorConfig)
throws Exception {
- Map<String, String> newProps = new HashMap<>(initialConfigMap);
- overrideProps(newProps, newDynamicConfigs);
- Configuration newConfig = Configuration.fromMap(newProps);
- Configuration oldConfig = currentConfig;
- Set<ServerReconfigurable> appliedServerReconfigurableSet = new HashSet<>();
- if (!newProps.equals(currentConfigMap)) {
- serverReconfigures
- .values()
- .forEach(
- serverReconfigurable -> {
- try {
- serverReconfigurable.validate(newConfig);
- } catch (ConfigException e) {
- LOG.error(
- "Validate new dynamic config error and will roll back all the applied config.",
- e);
- if (!skipErrorConfig) {
- throw e;
- }
- }
- });
-
- Exception throwable = null;
- for (ServerReconfigurable serverReconfigurable : serverReconfigures.values()) {
- try {
- serverReconfigurable.reconfigure(newConfig);
- appliedServerReconfigurableSet.add(serverReconfigurable);
- } catch (ConfigException e) {
- LOG.error(
- "Apply new dynamic error and will roll back all the applied config.",
- e);
- if (!skipErrorConfig) {
- throwable = e;
- break;
- }
- }
+ // Compute effective config changes (merge with initial configs)
+ Map<String, String> effectiveChanges =
+ computeEffectiveChanges(newDynamicConfigs, skipErrorConfig);
+
+ // Early return if no effective changes
+ if (effectiveChanges.isEmpty()) {
+ return;
Review Comment:
It would be helpful to log an info message when no changes are detected. This aids debugging, for example, when an `ALTER CONFIG` command appears to have no effect.
##########
fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java:
##########
@@ -115,68 +148,296 @@ boolean isAllowedConfig(String key) {
private void updateCurrentConfig(Map<String, String> newDynamicConfigs, boolean skipErrorConfig)
throws Exception {
- Map<String, String> newProps = new HashMap<>(initialConfigMap);
- overrideProps(newProps, newDynamicConfigs);
- Configuration newConfig = Configuration.fromMap(newProps);
- Configuration oldConfig = currentConfig;
- Set<ServerReconfigurable> appliedServerReconfigurableSet = new HashSet<>();
- if (!newProps.equals(currentConfigMap)) {
- serverReconfigures
- .values()
- .forEach(
- serverReconfigurable -> {
- try {
- serverReconfigurable.validate(newConfig);
- } catch (ConfigException e) {
- LOG.error(
- "Validate new dynamic config error and will roll back all the applied config.",
- e);
- if (!skipErrorConfig) {
- throw e;
- }
- }
- });
-
- Exception throwable = null;
- for (ServerReconfigurable serverReconfigurable : serverReconfigures.values()) {
- try {
- serverReconfigurable.reconfigure(newConfig);
- appliedServerReconfigurableSet.add(serverReconfigurable);
- } catch (ConfigException e) {
- LOG.error(
- "Apply new dynamic error and will roll back all the applied config.",
- e);
- if (!skipErrorConfig) {
- throwable = e;
- break;
- }
- }
+ // Compute effective config changes (merge with initial configs)
+ Map<String, String> effectiveChanges =
+ computeEffectiveChanges(newDynamicConfigs, skipErrorConfig);
+
+ // Early return if no effective changes
+ if (effectiveChanges.isEmpty()) {
+ return;
+ }
+
+ // Build new configuration by merging initial + dynamic configs
+ Map<String, String> newConfigMap = buildConfigMap(effectiveChanges);
+ Configuration newConfig = Configuration.fromMap(newConfigMap);
+
+ // Apply changes to all registered ServerReconfigurable instances
+ applyToServerReconfigurables(newConfig, skipErrorConfig);
+
+ // Update internal state
+ updateInternalState(newConfig, newConfigMap, newDynamicConfigs);
+ LOG.info("Dynamic configs changed: {}", newDynamicConfigs);
+ }
+
+ /**
+ * Computes effective config changes by validating new configs and handling deletions.
+ *
+ * @param newDynamicConfigs new dynamic configs from ZooKeeper
+ * @param skipErrorConfig whether to skip invalid configs
+ * @return map of config changes that passed validation
+ * @throws ConfigException if validation fails and skipErrorConfig is false
+ */
+ private Map<String, String> computeEffectiveChanges(
+ Map<String, String> newDynamicConfigs, boolean skipErrorConfig) throws ConfigException {
+ Map<String, String> effectiveChanges = new HashMap<>();
+ Set<String> skippedConfigs = new HashSet<>();
+
+ // Process deleted configs: restore to initial value or remove
+ processDeletions(newDynamicConfigs, effectiveChanges, skippedConfigs, skipErrorConfig);
+
+ // Process added/modified configs
+ processModifications(newDynamicConfigs, effectiveChanges, skippedConfigs, skipErrorConfig);
+
+ if (!skippedConfigs.isEmpty()) {
+ LOG.warn("Skipped invalid configs: {}", skippedConfigs);
+ }
+
+ return effectiveChanges;
+ }
+
+ /** Processes config deletions by restoring to initial values or removing them. */
+ private void processDeletions(
+ Map<String, String> newDynamicConfigs,
+ Map<String, String> effectiveChanges,
+ Set<String> skippedConfigs,
+ boolean skipErrorConfig)
+ throws ConfigException {
+ for (String configKey : dynamicConfigs.keySet()) {
+ if (newDynamicConfigs.containsKey(configKey)) {
+ continue; // Not deleted
+ }
+
+ String currentValue = currentConfigMap.get(configKey);
+ String initialValue = initialConfigMap.get(configKey);
+
+ // Determine target value: initial value or null (removal)
+ String targetValue = initialValue;
+
+ // Skip if no change needed (already at initial value)
+ if (Objects.equals(currentValue, targetValue)) {
+ continue;
+ }
+
+ // Validate the change
+ if (validateConfigChange(
+ configKey, currentValue, targetValue, skippedConfigs, skipErrorConfig)) {
+ effectiveChanges.put(configKey, targetValue);
}
+ }
+ }
+
+ /** Processes config additions and modifications. */
+ private void processModifications(
+ Map<String, String> newDynamicConfigs,
+ Map<String, String> effectiveChanges,
+ Set<String> skippedConfigs,
+ boolean skipErrorConfig)
+ throws ConfigException {
+ for (Map.Entry<String, String> entry : newDynamicConfigs.entrySet()) {
+ String configKey = entry.getKey();
+ String newValue = entry.getValue();
+ String currentValue = currentConfigMap.get(configKey);
- // rollback to old config if there is an error.
- if (throwable != null) {
- appliedServerReconfigurableSet.forEach(
- serverReconfigurable -> serverReconfigurable.reconfigure(oldConfig));
- throw throwable;
+ // Skip if value unchanged
+ if (Objects.equals(currentValue, newValue)) {
+ continue;
}
- currentConfig = newConfig;
- currentConfigMap.clear();
- dynamicConfigs.clear();
- currentConfigMap.putAll(newProps);
- dynamicConfigs.putAll(newDynamicConfigs);
- LOG.info("Dynamic configs changed: {}", newDynamicConfigs);
+ // Validate and add to effective changes
+ if (validateConfigChange(
+ configKey, currentValue, newValue, skippedConfigs, skipErrorConfig)) {
+ effectiveChanges.put(configKey, newValue);
+ }
}
}
- private void overrideProps(Map<String, String> props, Map<String, String> propsOverride) {
- propsOverride.forEach(
+ /**
+ * Validates a single config change.
+ *
+ * @return true if validation passed, false if skipped due to error
+ * @throws ConfigException if validation fails and skipErrorConfig is false
+ */
+ private boolean validateConfigChange(
+ String configKey,
+ String oldValue,
+ String newValue,
+ Set<String> skippedConfigs,
+ boolean skipErrorConfig)
+ throws ConfigException {
+ try {
+ validateSingleConfig(configKey, oldValue, newValue);
+ return true;
+ } catch (ConfigException e) {
+ LOG.error(
+ "Config validation failed for '{}': {} -> {}. {}",
+ configKey,
+ oldValue,
+ newValue,
+ e.getMessage());
+ if (skipErrorConfig) {
+ skippedConfigs.add(configKey);
+ return false;
+ } else {
+ throw e;
+ }
+ }
+ }
+
+ /** Builds final config map by merging initial configs with effective changes. */
+ private Map<String, String> buildConfigMap(Map<String, String> effectiveChanges) {
+ Map<String, String> configMap = new HashMap<>(initialConfigMap);
+ effectiveChanges.forEach(
(key, value) -> {
if (value == null) {
- props.remove(key);
+ configMap.remove(key);
} else {
- props.put(key, value);
+ configMap.put(key, value);
}
});
+ return configMap;
+ }
+
+ /** Updates internal state after successful reconfiguration. */
+ private void updateInternalState(
+ Configuration newConfig,
+ Map<String, String> newConfigMap,
+ Map<String, String> newDynamicConfigs) {
+ currentConfig = newConfig;
+ currentConfigMap.clear();
+ currentConfigMap.putAll(newConfigMap);
+ dynamicConfigs.clear();
+ dynamicConfigs.putAll(newDynamicConfigs);
+ }
+
+ /**
+ * Validates a single config entry including type parsing and business validation.
+ *
+ * @param configKey config key
+ * @param oldValueStr old value string
+ * @param newValueStr new value string
+ * @throws ConfigException if validation fails
+ */
+ private void validateSingleConfig(String configKey, String oldValueStr, String newValueStr)
+ throws ConfigException {
+ // Get ConfigOption for type information
+ ConfigOption<?> configOption = ConfigOptions.getConfigOption(configKey);
+
+ // For configs with allowed prefixes (like "datalake."), skip ConfigOption validation
+ // and rely on ServerReconfigurable's business validation
+ boolean hasPrefixConfig = false;
+ for (String prefix : ALLOWED_CONFIG_PREFIXES) {
+ if (configKey.startsWith(prefix)) {
+ hasPrefixConfig = true;
+ break;
+ }
+ }
+
+ if (configOption == null && !hasPrefixConfig) {
+ throw new ConfigException(
+ String.format("No ConfigOption found for config key: %s", configKey));
+ }
+
+ // Parse and validate type only if ConfigOption exists
+ Object newValue = null;
+ if (configOption != null && newValueStr != null) {
+ Configuration tempConfig = new Configuration();
+ tempConfig.setString(configKey, newValueStr);
+ try {
+ newValue = tempConfig.getOptional(configOption).get();
+ } catch (Exception e) {
+ throw new ConfigException(
+ String.format(
+ "Cannot parse '%s' as %s for config '%s'",
+ newValueStr,
+ configOption.isList()
+ ? "List<" + configOption.getClazz().getSimpleName() + ">"
+ : configOption.getClazz().getSimpleName(),
+ configKey),
+ e);
Review Comment:
Add the exception message into `ConfigException` top error message, because `ApiException` will truncate the exception stack and will hidden the root cause.
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBResourceContainer.java:
##########
@@ -73,29 +74,41 @@ public class RocksDBResourceContainer implements AutoCloseable {
private final boolean enableStatistics;
+ /** The shared rate limiter for all RocksDB instances. */
+ private final RateLimiter sharedRateLimiter;
+
/** The handles to be closed when the container is closed. */
private final ArrayList<AutoCloseable> handlesToClose;
@VisibleForTesting
RocksDBResourceContainer() {
- this(new Configuration(), null, false);
+ this(new Configuration(), null, false, null);
}
public RocksDBResourceContainer(ReadableConfig configuration, @Nullable File instanceBasePath) {
- this(configuration, instanceBasePath, false);
+ this(configuration, instanceBasePath, false, null);
}
public RocksDBResourceContainer(
ReadableConfig configuration,
@Nullable File instanceBasePath,
boolean enableStatistics) {
+ this(configuration, instanceBasePath, enableStatistics, null);
Review Comment:
**Nit:** It would be better to use a default `RateLimiter` instance instead of `null`, and add a `checkNotNull` guard for the `sharedRateLimiter` parameter in the constructor.
Currently, many places still pass `null` for `sharedRateLimiter`. While this is currently used test-only, it risks leaking into production code and could introduce subtle bugs in the future. Using a default rate limiter would make the API safer and more robust.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support shared RocksDB rate limiter in Fluss [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong closed pull request #2178: Support shared RocksDB rate limiter in Fluss
URL: https://github.com/apache/fluss/pull/2178
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support shared RocksDB rate limiter in Fluss [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong merged PR #2178:
URL: https://github.com/apache/fluss/pull/2178
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support shared RocksDB rate limiter in Fluss [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #2178:
URL: https://github.com/apache/fluss/pull/2178#discussion_r2637686411
##########
website/docs/maintenance/operations/updating-configs.md:
##########
@@ -30,13 +34,50 @@ admin.alterClusterConfigs(
admin.alterClusterConfigs(
Collections.singletonList(
new AlterConfig(DATALAKE_FORMAT.key(), "paimon", AlterConfigOpType.DELETE)));
+
+// Set RocksDB shared rate limiter to 200MB/sec
+admin.alterClusterConfigs(
+ Collections.singletonList(
+ new AlterConfig("kv.shared-rate-limiter-bytes-per-sec", "209715200", AlterConfigOpType.SET)));
```
The `AlterConfig` class contains three properties:
* `key`: The configuration key to be modified (e.g., `datalake.format`)
* `value`: The configuration value to be set (e.g., `paimon`)
* `opType`: The operation type, either `AlterConfigOpType.SET` or `AlterConfigOpType.DELETE`
+### Using Flink Stored Procedures
+
+For certain configurations, Fluss provides convenient Flink stored procedures that can be called directly from Flink SQL.
+
+#### Managing RocksDB Rate Limiter
+
+You can dynamically adjust the shared RocksDB rate limiter for all TabletServers using Flink stored procedures:
+
+**Set Shared RocksDB Rate Limiter:**
+```sql title="Flink SQL"
+-- Set rate limiter to 200MB/sec (recommended, use named argument, only supported since Flink 1.19)
+CALL fluss_catalog.sys.set_shared_rocksdb_rate_limiter(rate_limit => '200MB');
+
+-- Set rate limiter to 500MB/sec (use indexed argument)
+CALL fluss_catalog.sys.set_shared_rocksdb_rate_limiter('500MB');
Review Comment:
It seems this is not supported yet, should use `set_cluster_config` in documentation?
##########
website/docs/maintenance/operations/updating-configs.md:
##########
@@ -15,9 +15,13 @@ From Fluss version 0.8 onwards, some of the server configs can be updated withou
Currently, the supported dynamically updatable server configurations include:
- `datalake.format`: Enable lakehouse storage by specifying the lakehouse format, e.g., `paimon`, `iceberg`.
- Options with prefix `datalake.${datalake.format}`
+- `kv.shared-rate-limiter-bytes-per-sec`: Control RocksDB flush and compaction write rate shared across all RocksDB instances on the TabletServer.
-You can update the configuration of a cluster with [Java client](apis/java-client.md).
+You can update the configuration of a cluster with [Java client](apis/java-client.md) or [Flink Stored Procedures](#using-flink-stored-procedures).
Review Comment:
```suggestion
You can update the configuration of a cluster with [Java client](#using-java-client) or [Flink Stored Procedures](#using-flink-stored-procedures).
```
##########
website/docs/maintenance/configuration.md:
##########
@@ -156,6 +156,7 @@ during the Fluss cluster working.
| kv.rocksdb.use-bloom-filter | Boolean | true | If true, every newly created SST file will contain a Bloom filter. It is enabled by default.
|
| kv.rocksdb.bloom-filter.bits-per-key | Double | 10.0 | Bits per key that bloom filter will use, this only take effect when bloom filter is used. The default value is 10.0.
|
| kv.rocksdb.bloom-filter.block-based-mode | Boolean | false | If true, RocksDB will use block-based filter instead of full filter, this only take effect when bloom filter is used. The default value is `false`.
|
+| kv.shared-rate-limiter-bytes-per-sec | MemorySize | 0b | The bytes per second rate limit for RocksDB flush and compaction operations shared across all RocksDB instances on the TabletServer. A value of 0 (default) means rate limiting is disabled. This configuration can be updated dynamically without server restart. See [Updating Configs](operations/updating-configs.md) for more details.
|
Review Comment:
Update the config key and the default value
##########
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)
Review Comment:
ditto
##########
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');
Review Comment:
Replace all the `kv.shared-rate-limiter.bytes-per-sec` in docs/javadocs to `kv.rocksdb.shared-rate-limiter.bytes-per-sec` which is the true config key.
##########
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)};
Review Comment:
If there is no configuration, should just return empty as is? Just like select from an empty table? Otherwise, this may be confusing there is a row in the config result:
config_key | config_value | config_source
-----|-----|-----
No cluster configurations found | |
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java:
##########
@@ -214,10 +246,13 @@ public static KvTablet create(
schemaGetter);
}
- private static RocksDBKv buildRocksDBKv(Configuration configuration, File kvDir)
+ private static RocksDBKv buildRocksDBKv(
+ Configuration configuration,
+ File kvDir,
+ @Nullable org.rocksdb.RateLimiter sharedRateLimiter)
Review Comment:
Can we replace the qualified class name with `import`?
##########
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();
Review Comment:
Could you please add a **“Procedures”** section under the **“DDL”** section in the **“Engine Flink”** documentation to list all supported procedures?
This can be tracked in a separate issue if preferred.
For the format, I’d recommend following the style used by Iceberg, using **headers and descriptive text** rather than tables.
Examples:
- https://iceberg.apache.org/docs/nightly/spark-procedures/#spark-procedures
- https://paimon.apache.org/docs/master/flink/procedures/
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvConfigValidator.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.server.kv;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.MemorySize;
+import org.apache.fluss.config.cluster.ConfigValidator;
+import org.apache.fluss.exception.ConfigException;
+
+import javax.annotation.Nullable;
+
+/**
+ * Stateless validator for RocksDB shared rate limiter configuration.
+ *
+ * <p>This validator can be registered on CoordinatorServer without requiring a KvManager instance,
+ * enabling early validation of KV configs before they are persisted to ZooKeeper.
+ *
+ * <p>The validator enforces the following rules for RocksDB shared rate limiter:
+ *
+ * <ul>
+ * <li>Cannot enable rate limiter dynamically if it was not enabled at server startup
+ * <li>Cannot disable rate limiter dynamically if it was enabled at server startup
Review Comment:
I don’t fully understand the reason behind this limitation. Is there significant overhead in always having a `RateLimiter` enabled in RocksDB, even when configured with a very high throughput limit (e.g., 1 TB/s)?
If the performance cost is negligible and especially since we’re enabling the rate limiter by default, I’d suggest treating a very high value (or `Long.MAX_VALUE`) as the effective “disabled” state (this also means we don't rely on `0` to disable)
I think switching between enable and disable is a common and useful functionality. Otherwise, it's very confusing the documentation says set to `0` to disable, but it doesn't work.
##########
website/docs/maintenance/operations/updating-configs.md:
##########
@@ -30,13 +34,50 @@ admin.alterClusterConfigs(
admin.alterClusterConfigs(
Collections.singletonList(
new AlterConfig(DATALAKE_FORMAT.key(), "paimon", AlterConfigOpType.DELETE)));
+
+// Set RocksDB shared rate limiter to 200MB/sec
+admin.alterClusterConfigs(
+ Collections.singletonList(
+ new AlterConfig("kv.shared-rate-limiter-bytes-per-sec", "209715200", AlterConfigOpType.SET)));
```
The `AlterConfig` class contains three properties:
* `key`: The configuration key to be modified (e.g., `datalake.format`)
* `value`: The configuration value to be set (e.g., `paimon`)
* `opType`: The operation type, either `AlterConfigOpType.SET` or `AlterConfigOpType.DELETE`
+### Using Flink Stored Procedures
+
+For certain configurations, Fluss provides convenient Flink stored procedures that can be called directly from Flink SQL.
+
+#### Managing RocksDB Rate Limiter
+
+You can dynamically adjust the shared RocksDB rate limiter for all TabletServers using Flink stored procedures:
+
+**Set Shared RocksDB Rate Limiter:**
+```sql title="Flink SQL"
+-- Set rate limiter to 200MB/sec (recommended, use named argument, only supported since Flink 1.19)
+CALL fluss_catalog.sys.set_shared_rocksdb_rate_limiter(rate_limit => '200MB');
Review Comment:
Invoke `USE fluss_catalog` before all the `CALL` statements, this makes the CALL statements is more reusable as the catalog name maybe different for different users.
##########
website/docs/maintenance/operations/updating-configs.md:
##########
@@ -30,13 +34,50 @@ admin.alterClusterConfigs(
admin.alterClusterConfigs(
Collections.singletonList(
new AlterConfig(DATALAKE_FORMAT.key(), "paimon", AlterConfigOpType.DELETE)));
+
+// Set RocksDB shared rate limiter to 200MB/sec
+admin.alterClusterConfigs(
+ Collections.singletonList(
+ new AlterConfig("kv.shared-rate-limiter-bytes-per-sec", "209715200", AlterConfigOpType.SET)));
Review Comment:
```suggestion
new AlterConfig(KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key(), "200MB", AlterConfigOpType.SET)));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org