You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "zuston (via GitHub)" <gi...@apache.org> on 2026/08/18 10:23:17 UTC
[PR] [server] Add support of dynamic historical lookup threads and max queued requests reconfiguration [fluss]
zuston opened a new pull request, #4034:
URL: https://github.com/apache/fluss/pull/4034
### Purpose
Historical partition lookup traffic is difficult to predict accurately in production.
Allowing the lookup thread pool size and maximum number of queued requests to be updated dynamically enables the system to adapt quickly to changing QPS without restarting the server.
### 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] [server] Add support of dynamic historical lookup threads and max queued requests reconfiguration [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #4034:
URL: https://github.com/apache/fluss/pull/4034#discussion_r3894062642
##########
website/docs/maintenance/configuration.md:
##########
@@ -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. |
Review Comment:
Do these permits cover historical writes too (HistoricalPartitionManager#putKv), not just lookups?
##########
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(
Review Comment:
This validator is only registered on the CoordinatorServer.
Should ReplicaManager#validate check these two keys too?
##########
fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionTaskExecutor.java:
##########
@@ -188,6 +195,32 @@ public int numInflightRequests() {
return maxQueuedHistoricalRequests - requestPermits.availablePermits();
}
+ /** Applies dynamic historical partition executor configuration changes. */
+ public synchronized void reconfigure(Configuration newConf) {
+ checkNotNull(newConf, "newConf must not be null.");
+ int newMaxQueuedHistoricalRequests =
+ newConf.get(ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS);
+ int newMaxThreadPoolSize =
+ newConf.get(ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE);
+ checkArgument(
Review Comment:
checkArgument throws IllegalArgumentException, but applyToServerReconfigurables only catches
ConfigException and so a bad value skips the rollback.
Can we throw ConfigException from a
validate() instead?
--
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] [server] Add support of dynamic historical lookup threads and max queued requests reconfiguration [fluss]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #4034:
URL: https://github.com/apache/fluss/pull/4034#issuecomment-5462996740
Could you help review this? @fresh-borzoni
--
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] [server] Add support of dynamic historical lookup threads and max queued requests reconfiguration [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #4034:
URL: https://github.com/apache/fluss/pull/4034#discussion_r3901280037
##########
fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionTaskExecutor.java:
##########
@@ -188,6 +195,24 @@ public int numInflightRequests() {
return maxQueuedHistoricalRequests - requestPermits.availablePermits();
}
+ /** Applies dynamic historical partition executor configuration changes. */
+ public synchronized void reconfigure(Configuration newConf) {
Review Comment:
reconfigure() has no guard of its own now, and validate() only stops it when skipErrorConfig is false. Not reachable while the coordinator rejects the value first, but a 0 on the queue would silently throttle everything rather than fail. A ConfigException mb?
--
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] [server] Add support of dynamic historical lookup threads and max queued requests reconfiguration [fluss]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #4034:
URL: https://github.com/apache/fluss/pull/4034#issuecomment-5597155065
ping @fresh-borzoni @luoyuxia
--
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] [server] Add support of dynamic historical lookup threads and max queued requests reconfiguration [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #4034:
URL: https://github.com/apache/fluss/pull/4034#discussion_r3904043932
##########
fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java:
##########
@@ -179,6 +181,24 @@ class ReplicaManagerTest extends ReplicaTestBase {
/** First PUT_KV version that understands the STORAGE_BACKPRESSURE_EXCEPTION error code. */
private static final short PUT_KV_VERSION_WITH_STORAGE_BACKPRESSURE = 2;
+ @Test
+ void testRejectNonPositiveHistoricalPartitionConfigs() {
Review Comment:
I think we can remove this test for simplicity. Or add similar test in HistoricalPartitionTaskExecutorTest to test reconfigure() reject non-positive value
##########
fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionTaskExecutor.java:
##########
@@ -59,12 +59,14 @@ public final class HistoricalPartitionTaskExecutor implements AutoCloseable {
private static final Duration SHUTDOWN_TIMEOUT = Duration.ofSeconds(10);
private static final String THREAD_NAME_PREFIX = "historical-partition-io";
- private final int maxQueuedHistoricalRequests;
+ private volatile int maxQueuedHistoricalRequests;
// Shared by lookup and write tasks, including tasks waiting in the executor queue.
- private final Semaphore requestPermits;
+ private final AdjustableSemaphore requestPermits;
// Keep accepted requests so close() can cancel work that remains after executor shutdown.
private final Set<CompletableFuture<?>> pendingRequests;
private final ExecutorService executor;
+ private final @Nullable ThreadPoolExecutor threadPoolExecutor;
Review Comment:
I think we don't need to keep it as the class variable. A nullable threadPoolExecutor make me feel a little of confused to figure out when it'll be null. We can just
```
private void resizeThreadPool(int newMaxThreadPoolSize) {
if (!(executor instanceof ThreadPoolExecutor)) {
throw new IllegalStateException(
"Historical partition executor does not support dynamic resizing.");
}
ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executor;
int currentMaxThreadPoolSize = threadPoolExecutor.getMaximumPoolSize();
if (newMaxThreadPoolSize > currentMaxThreadPoolSize) {
threadPoolExecutor.setMaximumPoolSize(newMaxThreadPoolSize);
threadPoolExecutor.setCorePoolSize(newMaxThreadPoolSize);
} else if (newMaxThreadPoolSize < currentMaxThreadPoolSize) {
threadPoolExecutor.setCorePoolSize(newMaxThreadPoolSize);
threadPoolExecutor.setMaximumPoolSize(newMaxThreadPoolSize);
}
}
```
##########
fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionTaskExecutor.java:
##########
@@ -207,4 +232,40 @@ private static ExecutorService createHistoricalPartitionExecutor(int maxThreadPo
executor.allowCoreThreadTimeOut(true);
return executor;
}
+
+ private void resizeThreadPool(int newMaxThreadPoolSize) {
+ if (threadPoolExecutor == null) {
+ throw new IllegalStateException(
+ "Historical partition executor does not support dynamic resizing.");
+ }
+ int currentMaxThreadPoolSize = threadPoolExecutor.getMaximumPoolSize();
+ if (newMaxThreadPoolSize > currentMaxThreadPoolSize) {
+ threadPoolExecutor.setMaximumPoolSize(newMaxThreadPoolSize);
+ threadPoolExecutor.setCorePoolSize(newMaxThreadPoolSize);
+ } else if (newMaxThreadPoolSize < currentMaxThreadPoolSize) {
+ threadPoolExecutor.setCorePoolSize(newMaxThreadPoolSize);
+ threadPoolExecutor.setMaximumPoolSize(newMaxThreadPoolSize);
+ }
+ }
+
+ private void adjustRequestPermits(int newMaxQueuedHistoricalRequests) {
+ int permitDelta = newMaxQueuedHistoricalRequests - maxQueuedHistoricalRequests;
+ if (permitDelta > 0) {
+ requestPermits.release(permitDelta);
+ } else if (permitDelta < 0) {
+ requestPermits.decreasePermits(-permitDelta);
+ }
+ }
+
+ private static final class AdjustableSemaphore extends Semaphore {
+ private static final long serialVersionUID = 1L;
+
+ private AdjustableSemaphore(int permits) {
+ super(permits);
+ }
+
+ private void decreasePermits(int reduction) {
Review Comment:
nit: maybe can simplify to
```
private static final class AdjustableSemaphore extends Semaphore {
private static final long serialVersionUID = 1L;
private AdjustableSemaphore(int permits) {
super(permits);
}
private void adjustPermits(int delta) {
if (delta > 0) {
release(delta);
} else if (delta < 0) {
reducePermits(-delta);
}
}
}
```
--
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] [server] Add support of dynamic historical lookup threads and max queued requests reconfiguration [fluss]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #4034:
URL: https://github.com/apache/fluss/pull/4034#issuecomment-5611784882
thanks @fresh-borzoni @luoyuxia .
--
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] [server] Add support of dynamic historical lookup threads and max queued requests reconfiguration [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #4034:
URL: https://github.com/apache/fluss/pull/4034#discussion_r3894028123
##########
fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionTaskExecutor.java:
##########
@@ -188,6 +195,32 @@ public int numInflightRequests() {
return maxQueuedHistoricalRequests - requestPermits.availablePermits();
}
+ /** Applies dynamic historical partition executor configuration changes. */
+ public synchronized void reconfigure(Configuration newConf) {
+ checkNotNull(newConf, "newConf must not be null.");
+ int newMaxQueuedHistoricalRequests =
+ newConf.get(ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS);
+ int newMaxThreadPoolSize =
+ newConf.get(ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE);
+ checkArgument(
Review Comment:
checkArgument throws IllegalArgumentException, but applyToServerReconfigurables only catches
ConfigException and so a bad value skips the rollback.
Can we throw ConfigException from a validate() instead?
--
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] [server] Add support of dynamic historical lookup threads and max queued requests reconfiguration [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni merged PR #4034:
URL: https://github.com/apache/fluss/pull/4034
--
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] [server] Add support of dynamic historical lookup threads and max queued requests reconfiguration [fluss]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #4034:
URL: https://github.com/apache/fluss/pull/4034#issuecomment-5564200778
cc @luoyuxia @fresh-borzoni
--
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] [server] Add support of dynamic historical lookup threads and max queued requests reconfiguration [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #4034:
URL: https://github.com/apache/fluss/pull/4034#issuecomment-5602406301
Will merge after ci pass
--
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] [server] Add support of dynamic historical lookup threads and max queued requests reconfiguration [fluss]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #4034:
URL: https://github.com/apache/fluss/pull/4034#discussion_r3901366781
##########
fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionTaskExecutor.java:
##########
@@ -188,6 +195,24 @@ public int numInflightRequests() {
return maxQueuedHistoricalRequests - requestPermits.availablePermits();
}
+ /** Applies dynamic historical partition executor configuration changes. */
+ public synchronized void reconfigure(Configuration newConf) {
Review Comment:
Good point! This is a common use case with the way config options are currently handled. From a design perspective, we could introduce a `valueValidator` at the config-option level. A cleaner approach might look like this:
```
public static final ConfigOption<Integer> KV_RECOVERY_REMOTE_LOG_PREFETCH_NUM =
key("kv.recover.remote-log.prefetch-num")
.intType()
.defaultValue(4)
.valueValidator(key -> key > 0)
.withDescription("xxxxx");
```
--
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