From ca3c940bb1ac475adaee7f5228300a58dac40def Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Tue, 18 Aug 2026 18:14:44 +0800 Subject: [PATCH] [server] Add support of dynamic historical lookup threads and max queued requests reconfiguration --- .../fluss/server/DynamicServerConfig.java | 4 + .../HistoricalLookupCacheConfigValidator.java | 17 ++- .../replica/HistoricalLakeLookupManager.java | 102 +++++++++++++-- .../fluss/server/replica/ReplicaManager.java | 1 + .../fluss/server/DynamicConfigChangeTest.java | 24 ++++ .../HistoricalLakeLookupManagerTest.java | 118 +++++++++++++++++- website/docs/maintenance/configuration.md | 6 +- .../operations/updating-configs.md | 2 + 8 files changed, 259 insertions(+), 15 deletions(-) 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 index 98d721e152e..cac5e248ca3 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java @@ -51,6 +51,7 @@ import static org.apache.fluss.config.ConfigOptions.KV_SNAPSHOT_INTERVAL; import static org.apache.fluss.config.ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER; import static org.apache.fluss.config.ConfigOptions.LOG_RETENTION_ROLL_ACTIVE_SEGMENT_ENABLED; +import static org.apache.fluss.config.ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS; import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS; import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS_STRATEGY; import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS; @@ -58,6 +59,7 @@ import static org.apache.fluss.config.ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO; import static org.apache.fluss.config.ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS; import static org.apache.fluss.config.ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO; +import static org.apache.fluss.config.ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE; import static org.apache.fluss.config.ConfigOptions.SERVER_SASL_CREDENTIALS; import static org.apache.fluss.config.ConfigOptions.SERVER_SASL_PLAIN_JAAS_CONFIG; import static org.apache.fluss.utils.concurrent.LockUtils.inReadLock; @@ -86,6 +88,8 @@ class DynamicServerConfig { SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(), SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO.key(), SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS.key(), + SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE.key(), + NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key(), // Config options for remote.data.dirs REMOTE_DATA_DIRS.key(), REMOTE_DATA_DIRS_STRATEGY.key(), diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/HistoricalLookupCacheConfigValidator.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/HistoricalLookupCacheConfigValidator.java index c9d763f410a..dce63622daa 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/HistoricalLookupCacheConfigValidator.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/HistoricalLookupCacheConfigValidator.java @@ -24,11 +24,18 @@ import java.time.Duration; -/** Validates dynamic historical lookup cache settings. */ +/** Validates dynamic historical lookup settings. */ final class HistoricalLookupCacheConfigValidator implements ServerReconfigurable { @Override public void validate(Configuration newConfig) throws ConfigException { + validatePositive( + newConfig.get(ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE), + ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE.key()); + validatePositive( + newConfig.get(ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS), + ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key()); + double newMaxRatio = newConfig.get( ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO); @@ -56,4 +63,12 @@ public void validate(Configuration newConfig) throws ConfigException { @Override public void reconfigure(Configuration newConfig) {} + + private static void validatePositive(int value, String configKey) throws ConfigException { + if (value <= 0) { + throw new ConfigException( + String.format( + "Invalid configuration for %s, it must be greater than 0.", configKey)); + } + } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java index 977b608402b..e2f02b1ff0e 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java @@ -21,6 +21,7 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.config.TableConfig; +import org.apache.fluss.exception.ConfigException; import org.apache.fluss.exception.FlussRuntimeException; import org.apache.fluss.exception.HistoricalPartitionThrottledException; import org.apache.fluss.exception.InvalidPartitionException; @@ -71,7 +72,6 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutorService; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.Semaphore; import java.util.concurrent.ThreadPoolExecutor; @@ -131,12 +131,13 @@ class HistoricalLakeLookupManager implements AutoCloseable { private volatile long lakeConfigVersion; private final @Nullable PluginManager pluginManager; private final Counter capacityEvictions; - private final int maxQueuedHistoricalRequests; - private final Semaphore lookupPermits; + private volatile int maxQueuedHistoricalRequests; + private final AdjustableSemaphore lookupPermits; // Accepted lookup futures tracked so close() can cancel tasks left after executor shutdown. private final Set> pendingLookups; private final Cache lakeTableLookupers; - private final ExecutorService historicalPartitionExecutor; + private final ThreadPoolExecutor historicalPartitionExecutor; + private volatile int maxThreadPoolSize; private final File historicalLookupCacheRootDir; private final long dataDirVolumeBytes; // TODO: Introduce a minimum lookup cache disk ratio (default 0.01). When disk usage is high, @@ -173,7 +174,7 @@ class HistoricalLakeLookupManager implements AutoCloseable { HistoricalLakeLookupManager( Configuration conf, @Nullable PluginManager pluginManager, - @Nullable ExecutorService historicalPartitionExecutor, + @Nullable ThreadPoolExecutor historicalPartitionExecutor, File dataDir, long dataDirVolumeBytes, Ticker ticker, @@ -209,6 +210,7 @@ class HistoricalLakeLookupManager implements AutoCloseable { historicalPartitionExecutor == null ? createHistoricalPartitionExecutor(maxThreadPoolSize) : historicalPartitionExecutor; + this.maxThreadPoolSize = maxThreadPoolSize; this.lakeTableLookupers = Caffeine.newBuilder() .maximumSize(MAX_CACHED_TABLES) @@ -221,7 +223,7 @@ class HistoricalLakeLookupManager implements AutoCloseable { .executor(Runnable::run) .removalListener(this::onLookuperRemoved) .build(); - this.lookupPermits = new Semaphore(maxQueuedHistoricalRequests); + this.lookupPermits = new AdjustableSemaphore(maxQueuedHistoricalRequests); this.pendingLookups = ConcurrentHashMap.newKeySet(); } @@ -341,7 +343,7 @@ private CompletableFuture submitLookup( return future; } - private ExecutorService createHistoricalPartitionExecutor(int maxThreadPoolSize) { + private ThreadPoolExecutor createHistoricalPartitionExecutor(int maxThreadPoolSize) { ThreadPoolExecutor executor = new ThreadPoolExecutor( maxThreadPoolSize, @@ -354,6 +356,18 @@ private ExecutorService createHistoricalPartitionExecutor(int maxThreadPoolSize) return executor; } + private static final class AdjustableSemaphore extends Semaphore { + private static final long serialVersionUID = 1L; + + private AdjustableSemaphore(int permits) { + super(permits); + } + + private void decreasePermits(int reduction) { + super.reducePermits(reduction); + } + } + /** Invalidates the cached lake lookuper for the given table. */ void invalidateTableLookuper(long tableId) { lakeTableLookupers.invalidate(tableId); @@ -369,17 +383,37 @@ Counter capacityEvictions() { return capacityEvictions; } - /** Returns the number of accepted historical lookup requests that have not completed. */ + /** + * Returns the number of accepted historical lookup requests that have not completed. + * + *

This value is intended for monitoring and may be transient while requests complete or the + * request limit is reconfigured. + */ int numInflightRequests() { return maxQueuedHistoricalRequests - lookupPermits.availablePermits(); } + /** Validates the historical lookup settings that can be changed at runtime. */ + static void validateConfig(Configuration config) throws ConfigException { + validatePositive( + config.get(ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS), + ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key()); + validatePositive( + config.get(ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE), + ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE.key()); + } + /** Applies dynamic historical lookup configuration changes. */ - void reconfigure(Configuration newConf) { + void reconfigure(Configuration newConf) throws ConfigException { checkNotNull(newConf, "newConf must not be null."); + validateConfig(newConf); boolean lakeConfigChanged; boolean cacheLimitChanged; boolean expirationChanged; + int newMaxQueuedHistoricalRequests = + newConf.get(ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS); + int newMaxThreadPoolSize = + newConf.get(ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE); Duration newExpiration = newConf.get( ConfigOptions @@ -391,8 +425,28 @@ void reconfigure(Configuration newConf) { ConfigOptions .SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO)); cacheLimitChanged = newMaxBytesPerTable != lookupCacheMaxDiskBytesPerTable; - lookupCacheMaxDiskBytesPerTable = newMaxBytesPerTable; + if (newMaxThreadPoolSize != maxThreadPoolSize) { + resizeHistoricalPartitionThreadPool(newMaxThreadPoolSize); + LOG.info( + "{} reconfigured: {} -> {}", + ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE.key(), + maxThreadPoolSize, + newMaxThreadPoolSize); + maxThreadPoolSize = newMaxThreadPoolSize; + } + + if (newMaxQueuedHistoricalRequests != maxQueuedHistoricalRequests) { + adjustLookupPermits(newMaxQueuedHistoricalRequests); + LOG.info( + "{} reconfigured: {} -> {}", + ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key(), + maxQueuedHistoricalRequests, + newMaxQueuedHistoricalRequests); + maxQueuedHistoricalRequests = newMaxQueuedHistoricalRequests; + } + + lookupCacheMaxDiskBytesPerTable = newMaxBytesPerTable; lakeConfigChanged = hasLakeConfigChanged(conf, newConf); expirationChanged = !newExpiration.equals( @@ -424,6 +478,34 @@ void reconfigure(Configuration newConf) { } } + private static void validatePositive(int value, String configKey) throws ConfigException { + if (value <= 0) { + throw new ConfigException( + String.format( + "Invalid configuration for %s, it must be greater than 0.", configKey)); + } + } + + private void resizeHistoricalPartitionThreadPool(int newMaxThreadPoolSize) { + int currentMaxThreadPoolSize = historicalPartitionExecutor.getMaximumPoolSize(); + if (newMaxThreadPoolSize > currentMaxThreadPoolSize) { + historicalPartitionExecutor.setMaximumPoolSize(newMaxThreadPoolSize); + historicalPartitionExecutor.setCorePoolSize(newMaxThreadPoolSize); + } else if (newMaxThreadPoolSize < currentMaxThreadPoolSize) { + historicalPartitionExecutor.setCorePoolSize(newMaxThreadPoolSize); + historicalPartitionExecutor.setMaximumPoolSize(newMaxThreadPoolSize); + } + } + + private void adjustLookupPermits(int newMaxQueuedHistoricalRequests) { + int permitDelta = newMaxQueuedHistoricalRequests - maxQueuedHistoricalRequests; + if (permitDelta > 0) { + lookupPermits.release(permitDelta); + } else if (permitDelta < 0) { + lookupPermits.decreasePermits(-permitDelta); + } + } + private LookupResultForBucket lookupInternal( LookupDataForBucket lookupData, TableInfo tableInfo, diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java index 75b0f2ed1c3..f42d1438a76 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java @@ -414,6 +414,7 @@ public int getCoordinatorEpoch() { public void validate(Configuration newConfig) throws ConfigException { // Type validation is already handled by DynamicServerConfig. // Here we only do basic sanity checks. + HistoricalLakeLookupManager.validateConfig(newConfig); int newMinInSyncReplicas = newConfig.get(ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER); if (newMinInSyncReplicas <= 0) { diff --git a/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java index 5bc4a1fe8ce..5fbd8906ecd 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java @@ -429,6 +429,30 @@ void testReStartupContainsNoMatchedDynamicConfig() throws Exception { } } + @Test + void testHistoricalLookupConfigsCanBeChangedDynamically() throws Exception { + DynamicConfigManager dynamicConfigManager = createManager(new Configuration()); + dynamicConfigManager.startup(); + + dynamicConfigManager.alterConfigs( + Arrays.asList( + new AlterConfig( + ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE + .key(), + "12", + AlterConfigOpType.SET), + new AlterConfig( + ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key(), + "60", + AlterConfigOpType.SET))); + + assertThat(zookeeperClient.fetchEntityConfig()) + .containsEntry( + ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE.key(), "12") + .containsEntry( + ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key(), "60"); + } + @Test void testPreventInvalidConfig() throws Exception { // Test that generic type validation prevents invalid config values diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java index afa53e5516b..8d591299cee 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java @@ -50,13 +50,13 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; -import java.util.concurrent.AbstractExecutorService; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CompletableFuture; import java.util.concurrent.FutureTask; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; @@ -171,6 +171,116 @@ void testHistoricalLookupMaxQueuedRequestsUsesExplicitConfig() throws Exception assertThat(third.getError().error()).isEqualTo(Errors.HISTORICAL_PARTITION_THROTTLED); } + @Test + void testReconfiguresMaxQueuedHistoricalRequests() throws Exception { + ManualExecutor executor = new ManualExecutor(); + HistoricalLakeLookupManager manager = createManager(1, executor); + + CompletableFuture first = + manager.lookup( + lookupData(new TableBucket(PARTITION_TABLE_ID, 1L, 0)), + PARTITION_TABLE_INFO, + PARTITION_TABLE_INFO.getSchemaInfo(), + NO_OP_LOOKUP_METRIC_RECORDER); + + manager.reconfigure(conf(2)); + CompletableFuture second = + manager.lookup( + lookupData(new TableBucket(PARTITION_TABLE_ID, 2L, 0)), + PARTITION_TABLE_INFO, + PARTITION_TABLE_INFO.getSchemaInfo(), + NO_OP_LOOKUP_METRIC_RECORDER); + LookupResultForBucket third = + manager.lookup( + lookupData(new TableBucket(PARTITION_TABLE_ID, 3L, 0)), + PARTITION_TABLE_INFO, + PARTITION_TABLE_INFO.getSchemaInfo(), + NO_OP_LOOKUP_METRIC_RECORDER) + .get(1, TimeUnit.SECONDS); + + assertThat(first).isNotDone(); + assertThat(second).isNotDone(); + assertThat(manager.numInflightRequests()).isEqualTo(2); + assertThat(third.getError().error()).isEqualTo(Errors.HISTORICAL_PARTITION_THROTTLED); + + executor.runNext(); + executor.runNext(); + assertThat(first).isDone(); + assertThat(second).isDone(); + assertThat(manager.numInflightRequests()).isZero(); + } + + @Test + void testReducesMaxQueuedHistoricalRequestsWhileLookupsAreInflight() throws Exception { + ManualExecutor executor = new ManualExecutor(); + HistoricalLakeLookupManager manager = createManager(2, executor); + + CompletableFuture first = + manager.lookup( + lookupData(new TableBucket(PARTITION_TABLE_ID, 1L, 0)), + PARTITION_TABLE_INFO, + PARTITION_TABLE_INFO.getSchemaInfo(), + NO_OP_LOOKUP_METRIC_RECORDER); + CompletableFuture second = + manager.lookup( + lookupData(new TableBucket(PARTITION_TABLE_ID, 2L, 0)), + PARTITION_TABLE_INFO, + PARTITION_TABLE_INFO.getSchemaInfo(), + NO_OP_LOOKUP_METRIC_RECORDER); + + manager.reconfigure(conf(1)); + LookupResultForBucket third = + manager.lookup( + lookupData(new TableBucket(PARTITION_TABLE_ID, 3L, 0)), + PARTITION_TABLE_INFO, + PARTITION_TABLE_INFO.getSchemaInfo(), + NO_OP_LOOKUP_METRIC_RECORDER) + .get(1, TimeUnit.SECONDS); + + assertThat(first).isNotDone(); + assertThat(second).isNotDone(); + assertThat(manager.numInflightRequests()).isEqualTo(2); + assertThat(third.getError().error()).isEqualTo(Errors.HISTORICAL_PARTITION_THROTTLED); + + executor.runNext(); + executor.runNext(); + assertThat(first).isDone(); + assertThat(second).isDone(); + assertThat(manager.numInflightRequests()).isZero(); + } + + @Test + void testReconfiguresHistoricalPartitionThreadPoolMaxSize() throws Exception { + Configuration initialConf = conf(1); + initialConf.set(ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE, 1); + ThreadPoolExecutor executor = + new ThreadPoolExecutor(1, 1, 1L, TimeUnit.MINUTES, new LinkedBlockingQueue<>()); + HistoricalLakeLookupManager manager = + new HistoricalLakeLookupManager( + initialConf, + null, + executor, + ioTmpDir, + DATA_DIR_VOLUME_BYTES, + Ticker.systemTicker(), + Scheduler.disabledScheduler(), + NO_OP_DISK_WRITE_GUARD); + + Configuration largerConf = new Configuration(initialConf); + largerConf.set(ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE, 3); + manager.reconfigure(largerConf); + assertThat(executor.getCorePoolSize()).isEqualTo(3); + assertThat(executor.getMaximumPoolSize()).isEqualTo(3); + + Configuration smallerConf = new Configuration(largerConf); + smallerConf.set(ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE, 1); + manager.reconfigure(smallerConf); + assertThat(executor.getCorePoolSize()).isEqualTo(1); + assertThat(executor.getMaximumPoolSize()).isEqualTo(1); + + manager.close(); + } + @Test void testRejectNonPositiveHistoricalLookupMaxQueuedRequests() { Configuration conf = conf(0); @@ -672,10 +782,14 @@ public ScheduledFuture schedule( } } - private static final class ManualExecutor extends AbstractExecutorService { + private static final class ManualExecutor extends ThreadPoolExecutor { private final BlockingQueue tasks = new LinkedBlockingQueue<>(); private volatile boolean shutdown; + private ManualExecutor() { + super(0, Integer.MAX_VALUE, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>()); + } + @Override public void shutdown() { shutdown = true; diff --git a/website/docs/maintenance/configuration.md b/website/docs/maintenance/configuration.md index b9e44b84069..e7c71493e84 100644 --- a/website/docs/maintenance/configuration.md +++ b/website/docs/maintenance/configuration.md @@ -21,8 +21,8 @@ auto-partition.check.interval: 5min ``` Server configuration refers to a set of configurations used to specify the running parameters of a server. -These settings can only be configured at the time of cluster startup and do not support dynamic modification -during the Fluss cluster working. +Most settings can only be configured at cluster startup. The configurations explicitly marked as dynamic +below can be modified while the Fluss cluster is running. ## Common @@ -101,6 +101,7 @@ The logging-related environment options (`env.log.dir`, `env.log.level`, `env.lo | server.data-disk.write-limit-ratio | Double | 0.85 | Reject writes when the tablet server data disk usage reaches this ratio. Writes resume when the usage reaches or drops below `server.data-disk.write-recover-ratio`. The monitor reports the maximum usage across all distinct file stores so that a single nearly-full disk is never masked by other low-usage disks. Set to `1.0` to disable the disk-usage protection entirely. The valid range is `(server.data-disk.write-recover-ratio, 1.0]`. When lowering both ratios dynamically, update them in the same request or lower `server.data-disk.write-recover-ratio` first. This configuration can be updated dynamically without server restart. | | server.data-disk.write-recover-ratio | Double | 0.80 | Resume writes when the tablet server data disk usage reaches or drops below this ratio. The valid range is `(0.0, server.data-disk.write-limit-ratio)`. This configuration can be updated dynamically without server restart. | | server.data-disk.check-interval | Duration | 30s | The interval at which the tablet server samples the local data disk usage for the write-protection state machine. A shorter interval narrows the time window during which writes can still flow in after the disk crosses the limit ratio, at the cost of slightly more frequent `statvfs` calls (which are in-memory and cheap). The default 30s is suitable for typical write workloads. | +| server.historical-partition.thread-pool.max-size | Integer | 10 | The maximum number of threads used for historical partition operations, such as lake lookups and writes. This configuration can be updated dynamically without server restart. | ## Zookeeper @@ -124,6 +125,7 @@ The logging-related environment options (`env.log.dir`, `env.log.level`, `env.lo | netty.server.num-network-threads | Integer | 3 | The number of threads that the server uses for receiving requests from the network and sending responses to the network. | | netty.server.num-worker-threads | Integer | 8 | The number of threads that the server uses for processing requests, which may include disk and remote I/O. | | netty.server.max-queued-requests | Integer | 500 | The number of queued requests allowed for worker threads, before blocking the I/O threads. | +| netty.server.max-queued-historical-requests | Integer | 50 | The number of historical lookup requests allowed to wait for lake lookup processing before throttling them. This configuration can be updated dynamically without server restart. | | netty.server.max-request-size | MemorySize | 100mb | The maximum size of a single request that the server can receive. This limits the maximum frame length at the Netty pipeline level to protect the server from malicious clients sending oversized requests that could exhaust server memory. | | netty.connection.max-idle-time | Duration | 10min | Close idle connections after the given time specified by this config. | | netty.client.num-network-threads | Integer | 4 | The number of threads that the client uses for sending requests to the network and receiving responses from network. The default value is 4. | diff --git a/website/docs/maintenance/operations/updating-configs.md b/website/docs/maintenance/operations/updating-configs.md index f6d3b12e38c..78506efdebd 100644 --- a/website/docs/maintenance/operations/updating-configs.md +++ b/website/docs/maintenance/operations/updating-configs.md @@ -21,6 +21,8 @@ Currently, the supported dynamically updatable server configurations include: - `datalake.format`: Specify the lakehouse format, e.g., `paimon`, `iceberg`. When enabling lakehouse storage explicitly, use it together with `datalake.enabled = true`. - Options with prefix `datalake.${datalake.format}` - `kv.rocksdb.shared-rate-limiter.bytes-per-sec`: Control RocksDB flush and compaction write rate shared across all RocksDB instances on the TabletServer. The rate limiter is always enabled. Set to a lower value (e.g., 100MB) to limit the rate, or a very high value to effectively disable rate limiting. +- `server.historical-partition.thread-pool.max-size`: Change the maximum number of threads used for historical partition operations. +- `netty.server.max-queued-historical-requests`: Change the maximum number of historical lookup requests admitted before throttling. You can update the configuration of a cluster with [Java client](#using-java-client) or [Flink SQL](#using-flink-sql).