refactor code & fix document link

platinumhamburg · platinumhamburg · commit e3f0c6ca4c02 · 2025-12-23T21:31:57.000+08:00
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java b/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java
@@ -154,6 +154,7 @@ private void updateCurrentConfig(Map<String, String> newDynamicConfigs, boolean
 
         // Early return if no effective changes
         if (effectiveChanges.isEmpty()) {
+            LOG.info("No effective config changes detected for: {}", newDynamicConfigs);
             return;
         }
 
@@ -166,7 +167,7 @@ private void updateCurrentConfig(Map<String, String> newDynamicConfigs, boolean
 
         // Update internal state
         updateInternalState(newConfig, newConfigMap, newDynamicConfigs);
-        LOG.info("Dynamic configs changed: {}", newDynamicConfigs);
+        LOG.info("Dynamic configs changed: {}", effectiveChanges);
     }
 
     /**
@@ -345,14 +346,17 @@ private void validateSingleConfig(String configKey, String oldValueStr, String n
             try {
                 newValue = tempConfig.getOptional(configOption).get();
             } catch (Exception e) {
+                String causeMessage =
+                        e.getMessage() != null ? e.getMessage() : e.getClass().getSimpleName();
                 throw new ConfigException(
                         String.format(
-                                "Cannot parse '%s' as %s for config '%s'",
+                                "Cannot parse '%s' as %s for config '%s': %s",
                                 newValueStr,
                                 configOption.isList()
                                         ? "List<" + configOption.getClazz().getSimpleName() + ">"
                                         : configOption.getClazz().getSimpleName(),
-                                configKey),
+                                configKey,
+                                causeMessage),
                         e);
             }
         }
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java
@@ -74,6 +74,39 @@
 public final class KvManager extends TabletManagerBase implements ServerReconfigurable {
 
     private static final Logger LOG = LoggerFactory.getLogger(KvManager.class);
+
+    /**
+     * Default global rate limiter with unlimited rate (Long.MAX_VALUE bytes per second).
+     *
+     * <p>This is used by RocksDBResourceContainer when no rate limiter is explicitly provided,
+     * ensuring the API is safer and more robust by avoiding null checks throughout the code.
+     */
+    private static final RateLimiter DEFAULT_RATE_LIMITER = createDefaultRateLimiter();
+
+    /**
+     * Creates a default rate limiter with unlimited rate (Long.MAX_VALUE bytes per second).
+     *
+     * @return a default rate limiter instance
+     */
+    private static RateLimiter createDefaultRateLimiter() {
+        RocksDB.loadLibrary();
+        // Create a rate limiter with unlimited rate (effectively no limit)
+        // Using default refill period and fairness values
+        return new RateLimiter(Long.MAX_VALUE);
+    }
+
+    /**
+     * Returns the default global rate limiter with unlimited rate.
+     *
+     * <p>This method provides access to the default rate limiter for use in
+     * RocksDBResourceContainer when no rate limiter is explicitly provided.
+     *
+     * @return the default rate limiter instance
+     */
+    public static RateLimiter getDefaultRateLimiter() {
+        return DEFAULT_RATE_LIMITER;
+    }
+
     private final LogManager logManager;
 
     private final TabletServerMetricGroup serverMetricGroup;
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBResourceContainer.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBResourceContainer.java
@@ -22,6 +22,7 @@
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.config.ReadableConfig;
+import org.apache.fluss.server.kv.KvManager;
 import org.apache.fluss.utils.FileUtils;
 import org.apache.fluss.utils.IOUtils;
 
@@ -49,6 +50,8 @@
 import java.util.Collection;
 import java.util.List;
 
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
 /* This file is based on source code of Apache Flink Project (https://flink.apache.org/), licensed by the Apache
  * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
  * additional information regarding copyright ownership. */
@@ -82,18 +85,18 @@ public class RocksDBResourceContainer implements AutoCloseable {
 
     @VisibleForTesting
     RocksDBResourceContainer() {
-        this(new Configuration(), null, false, null);
+        this(new Configuration(), null, false, KvManager.getDefaultRateLimiter());
     }
 
     public RocksDBResourceContainer(ReadableConfig configuration, @Nullable File instanceBasePath) {
-        this(configuration, instanceBasePath, false, null);
+        this(configuration, instanceBasePath, false, KvManager.getDefaultRateLimiter());
     }
 
     public RocksDBResourceContainer(
             ReadableConfig configuration,
             @Nullable File instanceBasePath,
             boolean enableStatistics) {
-        this(configuration, instanceBasePath, enableStatistics, null);
+        this(configuration, instanceBasePath, enableStatistics, KvManager.getDefaultRateLimiter());
     }
 
     public RocksDBResourceContainer(
@@ -108,7 +111,8 @@ public RocksDBResourceContainer(
                         ? RocksDBKvBuilder.getInstanceRocksDBPath(instanceBasePath)
                         : null;
         this.enableStatistics = enableStatistics;
-        this.sharedRateLimiter = sharedRateLimiter;
+        this.sharedRateLimiter =
+                checkNotNull(sharedRateLimiter, "sharedRateLimiter must not be null");
 
         this.handlesToClose = new ArrayList<>();
     }
@@ -130,10 +134,8 @@ public DBOptions getDbOptions() throws IOException {
         // add necessary default options
         opt = opt.setCreateIfMissing(true);
 
-        // set shared rate limiter if provided
-        if (sharedRateLimiter != null) {
-            opt.setRateLimiter(sharedRateLimiter);
-        }
+        // set shared rate limiter
+        opt.setRateLimiter(sharedRateLimiter);
 
         if (enableStatistics) {
             Statistics statistics = new Statistics();
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
@@ -192,7 +192,7 @@ private KvTablet createKvTablet(
                 DEFAULT_COMPRESSION,
                 schemaGetter,
                 tableConf.getChangelogImage(),
-                null);
+                KvManager.getDefaultRateLimiter());
     }
 
     @Test
diff --git a/website/docs/engine-flink/ddl/index.md b/website/docs/engine-flink/ddl/index.md
@@ -125,7 +125,7 @@ CREATE TABLE my_part_log_table (
 Fluss partitioned table supports dynamic partition creation, which means you can write data into a partition without pre-creating it.
 You can use the `INSERT INTO` statement to write data into a partitioned table, and Fluss will automatically create the partition if it does not exist.
 See the [Dynamic Partitioning](table-design/data-distribution/partitioning.md#dynamic-partitioning) for more details.
-But you can still use the [Add Partition](engine-flink/ddl.md#add-partition) statement to manually add partitions if needed.
+But you can still use the [Add Partition](engine-flink/ddl/index.md#add-partition) statement to manually add partitions if needed.
 :::
 
 #### Multi-Fields Partitioned Table
diff --git a/website/docs/engine-flink/getting-started.md b/website/docs/engine-flink/getting-started.md
@@ -22,19 +22,19 @@ For Flink's Table API, Fluss supports the following features:
 
 | Feature Support                                   | Flink | Notes                                  |
 |---------------------------------------------------|-------|----------------------------------------|
-| [SQL Create Catalog](ddl.md#create-catalog)       | ✔️    |                                        |
-| [SQL Create Database](ddl.md#create-database)     | ✔️    |                                        |
-| [SQL Drop Database](ddl.md#drop-database)         | ✔️    |                                        |
-| [SQL Create Table](ddl.md#create-table)           | ✔️    |                                        |
-| [SQL Create Table Like](ddl.md#create-table-like) | ✔️    |                                        |
-| [SQL Drop Table](ddl.md#drop-table)               | ✔️    |                                        |
-| [SQL Create Materialized Table](ddl.md#materialized-table) | ✔️    | Continuous refresh mode only |
-| [SQL Alter Materialized Table](ddl.md#alter-materialized-table) | ✔️    | Suspend/Resume support |
-| [SQL Drop Materialized Table](ddl.md#drop-materialized-table) | ✔️    |                                        |
-| [SQL Show Partitions](ddl.md#show-partitions)     | ✔️    |                                        |
-| [SQL Add Partition](ddl.md#add-partition)         | ✔️    |                                        |
-| [SQL Drop Partition](ddl.md#drop-partition)       | ✔️    |                                        |
-| [Procedures](ddl.md#procedures)                   | ✔️    | ACL management and cluster configuration |
+| [SQL Create Catalog](ddl/index.md#create-catalog)       | ✔️    |                                        |
+| [SQL Create Database](ddl/index.md#create-database)     | ✔️    |                                        |
+| [SQL Drop Database](ddl/index.md#drop-database)         | ✔️    |                                        |
+| [SQL Create Table](ddl/index.md#create-table)           | ✔️    |                                        |
+| [SQL Create Table Like](ddl/index.md#create-table-like) | ✔️    |                                        |
+| [SQL Drop Table](ddl/index.md#drop-table)               | ✔️    |                                        |
+| [SQL Create Materialized Table](ddl/index.md#materialized-table) | ✔️    | Continuous refresh mode only |
+| [SQL Alter Materialized Table](ddl/index.md#alter-materialized-table) | ✔️    | Suspend/Resume support |
+| [SQL Drop Materialized Table](ddl/index.md#drop-materialized-table) | ✔️    |                                        |
+| [SQL Show Partitions](ddl/index.md#show-partitions)     | ✔️    |                                        |
+| [SQL Add Partition](ddl/index.md#add-partition)         | ✔️    |                                        |
+| [SQL Drop Partition](ddl/index.md#drop-partition)       | ✔️    |                                        |
+| [Procedures](ddl/index.md#procedures)                   | ✔️    | ACL management and cluster configuration |
 | [SQL Select](reads.md)                            | ✔️    | Support both streaming and batch mode. |
 | [SQL Limit](reads.md#limit-read)                  | ✔️    | Only for Log Table                     |
 | [SQL Insert Into](writes.md)                      | ✔️    | Support both streaming and batch mode. |
diff --git a/website/docs/engine-flink/options.md b/website/docs/engine-flink/options.md
@@ -57,7 +57,7 @@ Using `ALTER TABLE ... SET` statement to modify the table options. For example,
 ALTER TABLE log_table SET ('table.datalake.enable' = 'true');
 ```
 
-See more details about [ALTER TABLE ... SET](engine-flink/ddl.md#set-properties) and [ALTER TABLE ... RESET](engine-flink/ddl.md#reset-properties) documentation.
+See more details about [ALTER TABLE ... SET](engine-flink/ddl/index.md#set-properties) and [ALTER TABLE ... RESET](engine-flink/ddl/index.md#reset-properties) documentation.
 
 ## Storage Options
 
diff --git a/website/docs/maintenance/operations/updating-configs.md b/website/docs/maintenance/operations/updating-configs.md
@@ -48,11 +48,11 @@ The `AlterConfig` class contains three properties:
 
 ### Using Flink Stored Procedures
 
-For certain configurations, Fluss provides convenient Flink stored procedures that can be called directly from Flink SQL. See [Procedures](../../engine-flink/ddl/procedures.md#cluster-configuration-procedures) for detailed documentation on using `get_cluster_config` and `set_cluster_config` procedures.
+For certain configurations, Fluss provides convenient Flink stored procedures that can be called directly from Flink SQL. See [Procedures](engine-flink/ddl/procedures.md#cluster-configuration-procedures) for detailed documentation on using `get_cluster_config` and `set_cluster_config` procedures.
 
 ## Updating Table Configs
 
-The connector options on a table including [Storage Options](engine-flink/options.md#storage-options) can be updated dynamically by [ALTER TABLE ... SET](engine-flink/ddl.md#alter-table) statement. See the example below:
+The connector options on a table including [Storage Options](engine-flink/options.md#storage-options) can be updated dynamically by [ALTER TABLE ... SET](engine-flink/ddl/index.md#alter-table) statement. See the example below:
 
 ```sql
 -- Enable lakehouse storage for the given table
diff --git a/website/docs/table-design/data-distribution/bucketing.md b/website/docs/table-design/data-distribution/bucketing.md
@@ -8,7 +8,7 @@ sidebar_position: 1
 A bucketing strategy is a data distribution technique that divides table data into small pieces 
 and distributes the data to multiple hosts and services.
 
-When creating a Fluss table, you can specify the number of buckets by setting `'bucket.num' = '<num>'` property for the table, see more details in [DDL](engine-flink/ddl.md).
+When creating a Fluss table, you can specify the number of buckets by setting `'bucket.num' = '<num>'` property for the table, see more details in [DDL](engine-flink/ddl/index.md).
 Currently, Fluss supports 3 bucketing strategies: **Hash Bucketing**, **Sticky Bucketing** and **Round-Robin Bucketing**.
 Primary-Key Tables only allow to use **Hash Bucketing**. Log Tables use **Sticky Bucketing** by default but can use other two bucketing strategies.
 
diff --git a/website/docs/table-design/data-distribution/partitioning.md b/website/docs/table-design/data-distribution/partitioning.md
@@ -9,7 +9,7 @@ sidebar_position: 2
 In Fluss, a **Partitioned Table** organizes data based on one or more partition keys, providing a way to improve query performance and manageability for large datasets. Partitions allow the system to divide data into distinct segments, each corresponding to specific values of the partition keys.
 
 For partitioned tables, Fluss supports three strategies of managing partitions.
-   - **Manual management partitions**, user can create new partitions or drop exists partitions. Learn how to create or drop partitions please refer to [Add Partition](engine-flink/ddl.md#add-partition) and [Drop Partition](engine-flink/ddl.md#drop-partition).
+   - **Manual management partitions**, user can create new partitions or drop exists partitions. Learn how to create or drop partitions please refer to [Add Partition](/engine-flink/ddl/index.md#add-partition) and [Drop Partition](engine-flink/ddl/index.md#drop-partition).
    - **Auto management partitions**, the partitions will be created based on the auto partitioning rules configured at the time of table creation, and expired partitions are automatically removed, ensuring data not expanding unlimited. See [Auto Partitioning](table-design/data-distribution/partitioning.md#auto-partitioning).
    - **Dynamic create partitions**, the partitions will be created automatically based on the data being written to the table. See [Dynamic Partitioning](table-design/data-distribution/partitioning.md#dynamic-partitioning).
    

Original file line number	Diff line number	Diff line change
`@@ -192,7 +192,7 @@ private KvTablet createKvTablet(`
`192`	`192`	`DEFAULT_COMPRESSION,`
`193`	`193`	`schemaGetter,`
`194`	`194`	`tableConf.getChangelogImage(),`
`195`		`- null);`
	`195`	`+ KvManager.getDefaultRateLimiter());`
`196`	`196`	`}`
`197`	`197`
`198`	`198`	`@Test`