You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "peterxcli (via GitHub)" <gi...@apache.org> on 2024/11/13 10:41:11 UTC
[PR] [WIP] KAFKA-17928: Make remote log manager thread-pool configs dynamic [kafka]
peterxcli opened a new pull request, #17793:
URL: https://github.com/apache/kafka/pull/17793
### Description
Add ability to dynamically update remote storage thread pool sizes and implement
partition readiness check:
- Add methods to RemoteLogManager to update thread pool sizes:
- updateCopyThreadPoolSize
- updateExpirationThreadPoolSize
- updateReaderThreadPoolSize
- Add thread pool configuration properties to DynamicRemoteLogConfig
- Add RLMM readiness check before executing remote storage tasks
- Update validation and reconfiguration logic in DynamicRemoteLogConfig
This change allows operators to tune thread pool sizes at runtime and prevents
operations on partitions that aren't ready for remote storage operations.
### Related Link
- [KIP-1105](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1105%3A+Make+remote+log+manager+thread-pool+configs+dynamic)
- [KAFKA-17928](https://issues.apache.org/jira/browse/KAFKA-17928?jql=project%20%3D%20KAFKA%20AND%20resolution%20%3D%20Unresolved%20AND%20assignee%20in%20(currentUser())%20ORDER%20BY%20priority%20DESC,%20updated%20DESC)
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17928: Make remote log manager thread-pool configs dynamic [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #17793:
URL: https://github.com/apache/kafka/pull/17793#issuecomment-2485331547
@peterxcli
There are 3 failed tests, could you fix them?
https://github.com/apache/kafka/actions/runs/11907042194/job/33180295070?pr=17793
```
FAILED ❌ KRaftClusterTest > testStartupWithNonDefaultKControllerDynamicConfiguration()
FAILED ❌ KRaftClusterTest > testIncreaseNumIoThreads()
FAILED ❌ DynamicBrokerReconfigurationTest > testThreadPoolResize(String).groupProtocol=classic
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-17928: Make remote log manager thread-pool configs dynamic [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #17793:
URL: https://github.com/apache/kafka/pull/17793#discussion_r1843560401
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -800,6 +815,11 @@ public void run() {
return;
}
+ if (!remoteLogMetadataManager.isReady(topicIdPartition)) {
Review Comment:
Why do we need this check? Could you rebase your branch?
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -1159,7 +1159,10 @@ class DynamicRemoteLogConfig(server: KafkaBroker) extends BrokerReconfigurable w
newConfig.values.forEach { (k, v) =>
if (RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP.equals(k) ||
RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP.equals(k) ||
- RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP.equals(k)) {
+ RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP.equals(k) ||
+ RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP.equals(k) ||
Review Comment:
could we move these 3 configs in a separate `if` check? This will help us in validating the change in thread count to be in the boundary (t/2 <= newSize <= t*2) similar to the validation done for other thread configs. Please refer to DynamicThreadPool.validateReconfiguration
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -289,6 +289,21 @@ public void updateFetchQuota(long quota) {
rlmFetchQuotaManager.updateQuota(new Quota(quota, true));
}
+ public void updateCopyThreadPoolSize(int size) {
+ LOGGER.info("Updating remote copy thread pool size to {}", size);
+ rlmCopyThreadPool.resize(size);
+ }
+
+ public void updateExpirationThreadPoolSize(int size) {
+ LOGGER.info("Updating remote expiration thread pool size to {}", size);
Review Comment:
shall we print the before and after size in the log? similar in other places.
```
Updating remote expiration thread pool size from {} to {}
```
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -1199,14 +1202,32 @@ class DynamicRemoteLogConfig(server: KafkaBroker) extends BrokerReconfigurable w
info(s"Dynamic remote log manager config: ${RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP} updated, " +
s"old value: $oldValue, new value: $newValue")
}
+ if (isChangedLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP)) {
+ val oldValue = oldLongValue(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP)
Review Comment:
this is not a long value, it should be int. We can check directly:
```
val newRLMConfig = newConfig.remoteLogManagerConfig
val oldRLMConfig = oldConfig.remoteLogManagerConfig
if (newRLMConfig.remoteLogManagerCopierThreadPoolSize() != oldRLMConfig.remoteLogManagerCopierThreadPoolSize())
...
``
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17928: Make remote log manager thread-pool configs dynamic [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #17793:
URL: https://github.com/apache/kafka/pull/17793#discussion_r1845022714
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -840,6 +847,43 @@ class BrokerDynamicThreadPool(server: KafkaBroker) extends BrokerReconfigurable
}
}
+
+class RemoteLogDynamicThreadPool(server: KafkaBroker) extends BrokerReconfigurable with Logging {
+ override def reconfigurableConfigs: Set[String] = Set(
+ RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP,
+ RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP,
+ RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP)
+
+ override def validateReconfiguration(newConfig: KafkaConfig): Unit = {
+ DynamicThreadPool.validateReconfiguration(server.config, newConfig)
+ }
+
+ override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
+ val remoteLogManager = server.remoteLogManagerOpt
+ if (newConfig.remoteLogCopierThreads != oldConfig.remoteLogCopierThreads) {
+ val oldValue = oldConfig.remoteLogCopierThreads
+ val newValue = newConfig.remoteLogCopierThreads
+ remoteLogManager.get.resizeCopierThreadPool(newValue)
+ info(s"Dynamic remote log manager config: ${RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP} updated, " +
+ s"old value: $oldValue, new value: $newValue")
+ }
+ if (newConfig.remoteLogExpirationThreads != oldConfig.remoteLogExpirationThreads) {
+ val oldValue = oldConfig.remoteLogExpirationThreads
+ val newValue = newConfig.remoteLogExpirationThreads
+ remoteLogManager.get.resizeExpirationThreadPool(newValue)
+ info(s"Dynamic remote log manager config: ${RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP} updated, " +
+ s"old value: $oldValue, new value: $newValue")
Review Comment:
same as above, we can avoid double log.
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -840,6 +847,43 @@ class BrokerDynamicThreadPool(server: KafkaBroker) extends BrokerReconfigurable
}
}
+
+class RemoteLogDynamicThreadPool(server: KafkaBroker) extends BrokerReconfigurable with Logging {
+ override def reconfigurableConfigs: Set[String] = Set(
+ RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP,
+ RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP,
+ RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP)
+
+ override def validateReconfiguration(newConfig: KafkaConfig): Unit = {
+ DynamicThreadPool.validateReconfiguration(server.config, newConfig)
+ }
+
+ override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
+ val remoteLogManager = server.remoteLogManagerOpt
+ if (newConfig.remoteLogCopierThreads != oldConfig.remoteLogCopierThreads) {
+ val oldValue = oldConfig.remoteLogCopierThreads
+ val newValue = newConfig.remoteLogCopierThreads
+ remoteLogManager.get.resizeCopierThreadPool(newValue)
+ info(s"Dynamic remote log manager config: ${RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP} updated, " +
+ s"old value: $oldValue, new value: $newValue")
+ }
+ if (newConfig.remoteLogExpirationThreads != oldConfig.remoteLogExpirationThreads) {
+ val oldValue = oldConfig.remoteLogExpirationThreads
+ val newValue = newConfig.remoteLogExpirationThreads
+ remoteLogManager.get.resizeExpirationThreadPool(newValue)
+ info(s"Dynamic remote log manager config: ${RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP} updated, " +
+ s"old value: $oldValue, new value: $newValue")
+ }
+ if (newConfig.remoteLogReaderThreads != oldConfig.remoteLogReaderThreads) {
+ val oldValue = oldConfig.remoteLogReaderThreads
+ val newValue = newConfig.remoteLogReaderThreads
+ remoteLogManager.get.resizeReaderThreadPool(newValue)
+ info(s"Dynamic remote log manager config: ${RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP} updated, " +
+ s"old value: $oldValue, new value: $newValue")
Review Comment:
same as above, we can avoid double log.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17928: Make remote log manager thread-pool configs dynamic [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #17793:
URL: https://github.com/apache/kafka/pull/17793#discussion_r1846672665
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -1219,6 +1264,6 @@ object DynamicRemoteLogConfig {
RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP,
RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP,
RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP,
- RemoteLogManagerConfig.REMOTE_LIST_OFFSETS_REQUEST_TIMEOUT_MS_PROP
+ RemoteLogManagerConfig.REMOTE_LIST_OFFSETS_REQUEST_TIMEOUT_MS_PROP,
Review Comment:
We have to remove the trailing comma, otherwise the build will fail:
See: https://github.com/apache/kafka/actions/runs/11892699554/job/33141799293
```
> Task :core:compileScala FAILED
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17928: Make remote log manager thread-pool configs dynamic [kafka]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli closed pull request #17793: KAFKA-17928: Make remote log manager thread-pool configs dynamic
URL: https://github.com/apache/kafka/pull/17793
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-17928: Make remote log manager thread-pool configs dynamic [kafka]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on PR #17793:
URL: https://github.com/apache/kafka/pull/17793#issuecomment-2480602052
Hi @kamalcph, Thanks for the review.
I do some refactors with your advice, PTAL, thanks! 😄
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17928: Make remote log manager thread-pool configs dynamic [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #17793:
URL: https://github.com/apache/kafka/pull/17793#discussion_r1845022036
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -840,6 +847,43 @@ class BrokerDynamicThreadPool(server: KafkaBroker) extends BrokerReconfigurable
}
}
+
+class RemoteLogDynamicThreadPool(server: KafkaBroker) extends BrokerReconfigurable with Logging {
+ override def reconfigurableConfigs: Set[String] = Set(
+ RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP,
+ RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP,
+ RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP)
+
+ override def validateReconfiguration(newConfig: KafkaConfig): Unit = {
+ DynamicThreadPool.validateReconfiguration(server.config, newConfig)
+ }
+
+ override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
+ val remoteLogManager = server.remoteLogManagerOpt
+ if (newConfig.remoteLogCopierThreads != oldConfig.remoteLogCopierThreads) {
+ val oldValue = oldConfig.remoteLogCopierThreads
+ val newValue = newConfig.remoteLogCopierThreads
+ remoteLogManager.get.resizeCopierThreadPool(newValue)
+ info(s"Dynamic remote log manager config: ${RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP} updated, " +
Review Comment:
The info log statement is not required as we already have a log inside `resizeCopierThreadPool`
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17928: Make remote log manager thread-pool configs dynamic [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #17793:
URL: https://github.com/apache/kafka/pull/17793#discussion_r1846695786
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -840,6 +849,42 @@ class BrokerDynamicThreadPool(server: KafkaBroker) extends BrokerReconfigurable
}
}
+
+class RemoteLogDynamicThreadPool(server: KafkaBroker) extends BrokerReconfigurable {
+
+ override def reconfigurableConfigs: Set[String] = Set(
+ RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP,
+ RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP,
+ RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP)
+
+ override def validateReconfiguration(newConfig: KafkaConfig): Unit = {
+ DynamicThreadPool.validateReconfiguration(server.config, newConfig)
+ }
+
+ override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
+ if (server.remoteLogManagerOpt.isEmpty) {
+ return
+ }
+
+ val remoteLogManager = server.remoteLogManagerOpt
Review Comment:
We can either rename the variable (or) get the value from the option:
```
val remoteLogManagerOpt = server.remoteLogManagerOpt
```
(or)
```
val remoteLogManager = server.remoteLogManagerOpt.get
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17928: Make remote log manager thread-pool configs dynamic [kafka]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on PR #17793:
URL: https://github.com/apache/kafka/pull/17793#issuecomment-2485797099
Hi @kamalcph
Should I close this PR?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17928: Make remote log manager thread-pool configs dynamic [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #17793:
URL: https://github.com/apache/kafka/pull/17793#discussion_r1845012900
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -771,7 +772,10 @@ object DynamicThreadPool {
ServerConfigs.NUM_IO_THREADS_CONFIG,
ReplicationConfigs.NUM_REPLICA_FETCHERS_CONFIG,
ServerLogConfigs.NUM_RECOVERY_THREADS_PER_DATA_DIR_CONFIG,
- ServerConfigs.BACKGROUND_THREADS_CONFIG)
+ ServerConfigs.BACKGROUND_THREADS_CONFIG,
+ RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP,
+ RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP,
+ RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP)
Review Comment:
We also have to update the BrokerDynamicThreadPool.reconfigurableConfigs() method to return only the below configs:
1. ServerConfigs.NUM_IO_THREADS_CONFIG
2. ReplicationConfigs.NUM_REPLICA_FETCHERS_CONFIG
3. ServerLogConfigs.NUM_RECOVERY_THREADS_PER_DATA_DIR_CONFIG and
4. ServerConfigs.BACKGROUND_THREADS_CONFIG
Otherwise for the newly added RemoteLogManager configs, it gets double registered as brokerReconfigurable in both BrokerDynamicThreadPool and RemoteLogDynamicThreadPool.
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -2163,6 +2178,10 @@ public RLMScheduledThreadPool(int poolSize, String threadPoolName, String thread
scheduledThreadPool = createPool();
}
+ public void resize(int newSize) {
Review Comment:
nit: Can we rename this method to `setCorePoolSize`? Also, add one getter method `getPoolSize`
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteStorageThreadPool.java:
##########
@@ -61,6 +61,10 @@ protected void afterExecute(Runnable runnable, Throwable th) {
}
}
+ public void resize(int newSize) {
Review Comment:
this method is not required, we can directly call `setCorePoolSize`.
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -840,6 +847,43 @@ class BrokerDynamicThreadPool(server: KafkaBroker) extends BrokerReconfigurable
}
}
+
+class RemoteLogDynamicThreadPool(server: KafkaBroker) extends BrokerReconfigurable with Logging {
+ override def reconfigurableConfigs: Set[String] = Set(
+ RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP,
+ RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP,
+ RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP)
+
+ override def validateReconfiguration(newConfig: KafkaConfig): Unit = {
+ DynamicThreadPool.validateReconfiguration(server.config, newConfig)
+ }
+
+ override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
+ val remoteLogManager = server.remoteLogManagerOpt
Review Comment:
Also, please add one if check as remoteLogManger is optional:
```suggestion
if (server.remoteLogManagerOpt.nonEmpty) {
val remoteLogManager = server.remoteLogManagerOpt.get
...
...
```
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -289,6 +289,21 @@ public void updateFetchQuota(long quota) {
rlmFetchQuotaManager.updateQuota(new Quota(quota, true));
}
+ public void resizeCopierThreadPool(int size) {
+ LOGGER.info("Updating remote copy thread pool size to {}", size);
+ rlmCopyThreadPool.resize(size);
+ }
Review Comment:
```suggestion
public void resizeCopierThreadPool(int newSize) {
int currentSize = rlmCopyThreadPool.getPoolSize();
LOGGER.info("Updating remote copy thread pool size from {} to {}", currentSize, newSize);
rlmCopyThreadPool.resize(newSize);
}
```
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -797,6 +801,9 @@ object DynamicThreadPool {
case ReplicationConfigs.NUM_REPLICA_FETCHERS_CONFIG => config.numReplicaFetchers
case ServerLogConfigs.NUM_RECOVERY_THREADS_PER_DATA_DIR_CONFIG => config.numRecoveryThreadsPerDataDir
case ServerConfigs.BACKGROUND_THREADS_CONFIG => config.backgroundThreads
+ case RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP => config.remoteLogCopierThreads
Review Comment:
Get the copier thread size using the below way, we don't want to expose the remoteLogManager configs in KafkaConfig class:
```suggestion
case RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP => config.remoteLogManagerConfig.remoteLogManagerCopierThreadPoolSize()
```
same for other configs
##########
core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala:
##########
@@ -1114,6 +1114,9 @@ class KafkaConfigTest {
case RemoteLogManagerConfig.REMOTE_LOG_READER_MAX_PENDING_TASKS_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1)
case RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", -3)
case RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", -3)
+ case RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1)
+ case RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1)
+ case RemoteLogManagerConfig.REMOTE_LIST_OFFSETS_REQUEST_TIMEOUT_MS_PROP => assertPropertyInvalid(baseProperties, name, "not_a_number", 0, -1)
Review Comment:
this is not required.
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -380,6 +380,9 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
def backgroundThreads = getInt(ServerConfigs.BACKGROUND_THREADS_CONFIG)
def numIoThreads = getInt(ServerConfigs.NUM_IO_THREADS_CONFIG)
+ def remoteLogCopierThreads = getInt(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP)
+ def remoteLogExpirationThreads = getInt(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP)
+ def remoteLogReaderThreads = getInt(RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP)
Review Comment:
These getters are not required, please see above comments.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17928: Make remote log manager thread-pool configs dynamic [kafka]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on PR #17793:
URL: https://github.com/apache/kafka/pull/17793#issuecomment-2482946541
@kamalcph Thanks for the detailed review, I made some correction base on it.
Please take a look ~
Thanks 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17928: Make remote log manager thread-pool configs dynamic [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #17793:
URL: https://github.com/apache/kafka/pull/17793#issuecomment-2485668397
@peterxcli
I've fixed the unit tests and opened #17859 PR. Added you as the co-author. Thought to open a PR against your branch but you've multiple `trunk` merges instead of `rebase`. PTAL.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17928: Make remote log manager thread-pool configs dynamic [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #17793:
URL: https://github.com/apache/kafka/pull/17793#issuecomment-2485837173
yes, we can close this PR and follow-up on #17859
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17928: Make remote log manager thread-pool configs dynamic [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #17793:
URL: https://github.com/apache/kafka/pull/17793#discussion_r1846706144
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -840,6 +849,42 @@ class BrokerDynamicThreadPool(server: KafkaBroker) extends BrokerReconfigurable
}
}
+
+class RemoteLogDynamicThreadPool(server: KafkaBroker) extends BrokerReconfigurable {
+
+ override def reconfigurableConfigs: Set[String] = Set(
+ RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP,
+ RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP,
+ RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP)
+
+ override def validateReconfiguration(newConfig: KafkaConfig): Unit = {
+ DynamicThreadPool.validateReconfiguration(server.config, newConfig)
+ }
+
+ override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
+ if (server.remoteLogManagerOpt.isEmpty) {
+ return
+ }
+
+ val remoteLogManager = server.remoteLogManagerOpt
+ if (newConfig.remoteLogCopierThreads != oldConfig.remoteLogCopierThreads) {
+ val oldValue = oldConfig.remoteLogCopierThreads
+ val newValue = newConfig.remoteLogCopierThreads
+ remoteLogManager.get.resizeCopierThreadPool(oldValue, newValue)
+ }
Review Comment:
Better to extract the remoteLogManagerConfig:
```
val remoteLogManager = server.remoteLogManagerOpt.get
val oldRLMConfig = oldConfig.remoteLogManagerConfig
val newRLMConfig = newConfig.remoteLogManagerConfig
if (oldRLMConfig.remoteLogManagerCopierThreadPoolSize() != newRLMConfig.remoteLogManagerCopierThreadPoolSize()) {
remoteLogManager.resizeCopierThreadPool(newRLMConfig.remoteLogManagerCopierThreadPoolSize())
}
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17928: Make remote log manager thread-pool configs dynamic [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #17793:
URL: https://github.com/apache/kafka/pull/17793#discussion_r1846671720
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -289,6 +289,21 @@ public void updateFetchQuota(long quota) {
rlmFetchQuotaManager.updateQuota(new Quota(quota, true));
}
+ public void resizeCopierThreadPool(int oldSize, int newSize) {
Review Comment:
Instead of passing the `oldSize`, can we get it from the threadPool itself?
```java
int currentSize = rlmCopyThreadPool.getCorePoolSize();
```
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -1219,6 +1264,6 @@ object DynamicRemoteLogConfig {
RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP,
RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP,
RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP,
- RemoteLogManagerConfig.REMOTE_LIST_OFFSETS_REQUEST_TIMEOUT_MS_PROP
+ RemoteLogManagerConfig.REMOTE_LIST_OFFSETS_REQUEST_TIMEOUT_MS_PROP,
Review Comment:
nit: we can remove this trailing comma.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org