Support intra-broker throttling (replica.alter.log.dirs.io.max.bytes.per.second) #2145

dongjinleekr · 2024-04-21T15:14:37Z

This PR resolves #1851.

dongjinleekr · 2024-04-21T15:18:19Z

This PR is an WIP; I will test this feature in our in-house fork (@naver) this week. Stay tuned!! 🙃

dongjinleekr · 2024-06-20T23:58:27Z

@aswinayyolath @mhratson @jiao-zhangS Could you kindly have a look? We adopted this patch to our in-house distribution and confirmed it works correctly. 🙏

cpaika

Will make things a lot more stable for JBOD clusters, good change

cpaika · 2024-08-15T15:36:31Z

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/executor/Executor.java

        inExecutionTasks = inExecutionTasks();
      }
+
+      if (_logDirThrottle != null) {


for my understanding, why the underscore in the variable names here?

In consistency with _replicationThrottle.

dongjinleekr · 2024-11-17T07:09:58Z

@cpaika @adkafka

Excuse me. Is there any issue with this PR? It initially worked on 2.5.138, and 2.5.141 was already released last month, but this PR has yet to be merged or released.

If any modifications are needed, don't hesitate to leave me a mention.

adkafka · 2024-11-18T16:28:09Z

@cpaika @adkafka

Excuse me. Is there any issue with this PR? It initially worked on 2.5.138, and 2.5.141 was already released last month, but this PR has yet to be merged or released.

If any modifications are needed, don't hesitate to leave me a mention.

I don't see any issues with this PR, but I'm not a maintainer of this project. We need a maintainer to review this and add their approval.

aswinayyolath · 2024-11-18T16:33:01Z

@CCisGG could you please review this PR?

dongjinleekr · 2024-12-07T13:15:12Z

@CCisGG Hello. Could you please have a look when you are free? 🙏

CCisGG · 2024-12-08T19:21:43Z

Hi @mhratson, would you mind taking a look at this one? Thanks!

imans777 · 2025-01-04T14:35:05Z

Anyone caring to review this?
It's really a must-have feature.

CCisGG

Added some comments/questions.

Some other questions I have:

Did you test that with your change, when no replication/intra broker throttling was set (which I think is by default), does this work properly?
Is their any case that this intra-broker throttling can interfere with inter-broker thorttling? E.g. what happens if they are enabled together?
Please update Configuration.md for this new config.

...-control/src/main/java/com/linkedin/kafka/cruisecontrol/config/constants/ExecutorConfig.java

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/executor/Executor.java

...ntrol/src/main/java/com/linkedin/kafka/cruisecontrol/executor/ReplicationThrottleHelper.java

dongjinleekr · 2025-02-23T14:18:06Z

@CCisGG

Sorry for being late. Here are the updates, rebased onto the latest main, and the anwsers:

Did you test that with your change, when no replication/intra broker throttling was set (which I think is by default), does this work properly?

YES. I tested it before I opened the PR, and we have been running it with our in-house distribution for more than 6 months and experienced no trouble. (As you already pointed out, default.log.dir.throttle is null by default.)

Is their any case that this intra-broker throttling can interfere with inter-broker thorttling? E.g. what happens if they are enabled together?

As far as I know, there is no case where intra and inter-broker throttlings interfere with each other. The reason is simple: the intra-broker rebalancing is only executed when rebalance_disk parameter is set to true. If this parameter is set to false (default), only inter-broker rebalancing is executed.

In short, intra- and inter-broker rebalancings are executed exclusively for each other, so if default.replication.throttle and default.log.dir.throttle are enabled together, they will not impact each other.

luciano-sabenca · 2025-03-18T09:56:16Z

Any news about this MR?
It's a really useful feature for us!
:D

CCisGG

Thanks for the change! I left some minor comments.

docs/wiki/User Guide/Configurations.md

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/KafkaCruiseControl.java

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/executor/Executor.java

...ntrol/src/main/java/com/linkedin/kafka/cruisecontrol/executor/ReplicationThrottleHelper.java

…ationThrottledRate

…cationThrottles

- Add: setLogDirThrottles, setLogDirThrottledRateIfNecessary - Rename: removeReplicationThrottledRateFromBroker → removeThrottledRatesFromBroker

…oker rebalanacing - Rename: ReplicationThrottleHelper.clearThrottles → clearInterBrokerThrottles - Add ReplicationThrottleHelper.clearIntraBrokerThrottles - Executor.intraBrokerMoveReplicas now calls ReplicationThrottleHelper.setLogDirThrottles, ReplicationThrottleHelper.clearIntraBrokerThrottles

…ateFromBroker

…roker movements vs. intra-broker movements)

…, and UpdateTopicConfigurationRunnable. (reason: they do not incur any intra-broker replica movements)

dongjinleekr · 2025-04-02T15:53:09Z

@CCisGG Rebased onto the latest main, with some additional improvements.

Let's create a task to track the split. Thanks!

You mean splitting the Executor#startExecution method call to inter-broker mode and intra-broker mode, applying replicationThrottle and logDirThrottle respectively, right? I tried to, but it seems impossible. Here is why:

The Executor#executeProposals method, which calls Executor#startExecution, is agnostic to the inter-broker and intra-broker modes.

(Note: The method call chain is KafkaCruiseControl#executeProposals → Executor#executeProposals → Executor#startExecution)

The rebalance_disk parameter (ParameterUtils.REBALANCE_DISK_MODE_PARAM) is only used to generate the proposal, and then the proposal is delivered to the Executor#executeProposals method to be executed. That is, the Executor#executeProposals method can not discriminate between the two modes.

Then, is splitting KafkaCruiseControl#executeProposals into inter- and intra-broker modes feasible? I also investigated the possibility, but it seems impossible. Here is why: it is called by the following 5 API calls/Runnables:

add_broker (AddBrokersRunnable)
fix_offline_replicas (FixOfflineReplicasRunnable)
rebalance (RebalanceRunnable)
RemoveDisksRunnable
UpdateTopicConfigurationRunnable

5 incurs no movements, 1 incurs inter-broker movements only, 3 incurs either inter-broker or intra-broker movements, and 2 and 4 incur BOTH. For this reason, splitting the two methods' calls seems unreasonable - we can not assert whether 2 or 4 invocations need intra-broker movements beforehand.

For the reasons above, I did nothing to the call chain. It is true that these methods seem too overloaded, but improving them with another PR may be much better IMHO. (I have an idea, and will open another PR on this issue after this PR is merged.)

dongjinleekr · 2025-04-02T15:56:21Z

@CCisGG Instead, I did:

removed logDirThrottle from the Executor#executeRemoval method.
Since its goal is to move all replicas in the specified broker, intra-broker replica movement does not occur at all. So, it is totally useless.
Remove _logDirThrottle from the AddBrokersRunnable, RemoveBrokersRunnable, and UpdateTopicConfigurationRunnable.
DITTO. These Runnables do not incur any intra-broker replica movements. So, it is also useless.

With one commit for Configurations.md, one commit for KafkaCruiseControl Javadoc, and two commits for the improvements above. In total, I added four commits after your last review. 🙏 Please have a look when you are free!! Thanks.

CCisGG · 2025-04-02T17:07:07Z

@dongjinleekr Thanks for the explanations and the new commits. I think they are all reasonable.

Since I probably won't have much time to do another round of review and audit, would you mind inviting some other person (maybe from your org) to review the PR? And Let's do a final round of testing to ensure this won't interfere with inter-broker rebalance. Once those are done, I can help to merge to PR. Let me know if this works. Thank you!

CCisGG

Looks good in general to me. Please ask for another review from a different person if possible. Then I can help to merge the PR.

dongjinleekr · 2025-04-03T11:13:21Z

@CCisGG

Would you mind inviting some other person (maybe from your org) to review the PR?

Oh no, no problem. In fact, I am now working on supporting this feature in the Strimzi Kafka operator, and it will be completed by this weekend. As soon as I open the PR, there may be other people who can review it. (Sadly, I am the only person who is working on CC internals in my org 😢)

Plus: We are running this feature in our in-house fork but, let me double-check if there is any problem 🙏 Thank you!

kyguy

Hi @dongjinleekr, the PR looks great overall! I just left a few minor suggestions, mostly related to refactoring for consistency and readability.

kyguy · 2025-10-03T17:48:45Z

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/executor/Executor.java

      LOG.info("User task {}: Starting {} intra-broker partition movements.", _uuid, numTotalPartitionMovements);

      int partitionsToMove = numTotalPartitionMovements;
+      Set<Integer> participatingBrokers = Sets.newHashSet();


Could we use new HashSet<>() from java.util instead?

It appears that Sets.newHashSet() is deprecated and uses java.util.HashSet under the hood anyway.

kyguy · 2025-10-03T20:02:35Z

...ntrol/src/main/java/com/linkedin/kafka/cruisecontrol/executor/ReplicationThrottleHelper.java

  }

-  void setThrottles(List<ExecutionProposal> replicaMovementProposals)
+  void setReplicationThrottles(List<ExecutionProposal> replicaMovementProposals, long throttleRate)


It looks like we refactor setThrottles() -> setReplicationThrottles() but then refactor clearThrottles() -> clearInterBrokerThrottles() instead of clearReplicationThrottles(). To improve readability, could we use a consistent naming scheme for the new and refactored method names in this class? For example, use:

(a) setReplicationThrottles() and clearReplicationThrottles()

or

(b) setInterBrokerThrottles() and clearInterBrokerThrottles()

Option (b) seems better aligned with Cruise Control terminology but either would be fine so long as it is consistent!

kyguy · 2025-10-03T20:02:59Z

...ntrol/src/main/java/com/linkedin/kafka/cruisecontrol/executor/ReplicationThrottleHelper.java

    }
  }

+  Set<Integer> setLogDirThrottles(List<ExecutionProposal> replicaMovementProposals, long throttleRate)


Similar comment as above

kyguy · 2025-10-03T20:05:48Z

...ntrol/src/main/java/com/linkedin/kafka/cruisecontrol/executor/ReplicationThrottleHelper.java

+    List<AlterConfigOp> ops = new ArrayList<>();
+    ConfigEntry currThrottleRate = brokerConfigs.get(LOG_DIR_THROTTLED_RATE);
+    if (currThrottleRate == null || !currThrottleRate.value().equals(String.valueOf(throttleRate))) {
+      LOG.debug("Setting {} to {} bytes/second for broker {}", LOG_DIR_THROTTLED_RATE, throttleRate, brokerId);
+      ops.add(new AlterConfigOp(new ConfigEntry(LOG_DIR_THROTTLED_RATE, String.valueOf(throttleRate)), AlterConfigOp.OpType.SET));
+    }
+    if (!ops.isEmpty()) {
+      changeBrokerConfigs(brokerId, ops);
+    }
+  }


Since there will only ever be one "ops", we could refactor to something like this:

Suggested change

List<AlterConfigOp> ops = new ArrayList<>();

ConfigEntry currThrottleRate = brokerConfigs.get(LOG_DIR_THROTTLED_RATE);

if (currThrottleRate == null || !currThrottleRate.value().equals(String.valueOf(throttleRate))) {

LOG.debug("Setting {} to {} bytes/second for broker {}", LOG_DIR_THROTTLED_RATE, throttleRate, brokerId);

ops.add(new AlterConfigOp(new ConfigEntry(LOG_DIR_THROTTLED_RATE, String.valueOf(throttleRate)), AlterConfigOp.OpType.SET));

}

if (!ops.isEmpty()) {

changeBrokerConfigs(brokerId, ops);

}

}

ConfigEntry currThrottleRate = brokerConfigs.get(LOG_DIR_THROTTLED_RATE);

if (currThrottleRate == null || !currThrottleRate.value().equals(String.valueOf(throttleRate))) {

LOG.debug("Setting {} to {} bytes/second for broker {}", LOG_DIR_THROTTLED_RATE, throttleRate, brokerId);

changeBrokerConfigs(brokerId, List.of(new AlterConfigOp(new ConfigEntry(LOG_DIR_THROTTLED_RATE, String.valueOf(throttleRate)), AlterConfigOp.OpType.SET)));

}

}

kyguy · 2025-10-03T20:09:25Z

...ntrol/src/main/java/com/linkedin/kafka/cruisecontrol/executor/ReplicationThrottleHelper.java

+    List<AlterConfigOp> ops = new ArrayList<>();
+    if (currLogDirThrottle != null) {
+      if (currLogDirThrottle.source().equals(ConfigEntry.ConfigSource.STATIC_BROKER_CONFIG)) {
+        LOG.debug("Skipping removal for static log dir throttle rate: {}", currLogDirThrottle);
+      } else {
+        LOG.debug("Removing log dir throttle rate: {} on broker {}", currLogDirThrottle, brokerId);
+        ops.add(new AlterConfigOp(new ConfigEntry(LOG_DIR_THROTTLED_RATE, null), AlterConfigOp.OpType.DELETE));
+      }
+    }
+    if (!ops.isEmpty()) {
+      changeBrokerConfigs(brokerId, ops);
+    }


Similar comment as above, since there is only ever one "ops", this could be refactored like the following:

Suggested change

List<AlterConfigOp> ops = new ArrayList<>();

if (currLogDirThrottle != null) {

if (currLogDirThrottle.source().equals(ConfigEntry.ConfigSource.STATIC_BROKER_CONFIG)) {

LOG.debug("Skipping removal for static log dir throttle rate: {}", currLogDirThrottle);

} else {

LOG.debug("Removing log dir throttle rate: {} on broker {}", currLogDirThrottle, brokerId);

ops.add(new AlterConfigOp(new ConfigEntry(LOG_DIR_THROTTLED_RATE, null), AlterConfigOp.OpType.DELETE));

}

}

if (!ops.isEmpty()) {

changeBrokerConfigs(brokerId, ops);

}

if (currLogDirThrottle != null) {

if (currLogDirThrottle.source().equals(ConfigEntry.ConfigSource.STATIC_BROKER_CONFIG)) {

LOG.debug("Skipping removal for static log dir throttle rate: {}", currLogDirThrottle);

} else {

LOG.debug("Removing log dir throttle rate: {} on broker {}", currLogDirThrottle, brokerId);

changeBrokerConfigs(brokerId, List.of(new AlterConfigOp(new ConfigEntry(LOG_DIR_THROTTLED_RATE, String.valueOf(throttleRate)), AlterConfigOp.OpType.SET)));

}

}

kyguy · 2025-10-03T20:23:27Z

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/executor/Executor.java

 import com.codahale.metrics.Gauge;
 import com.codahale.metrics.MetricRegistry;
 import com.codahale.metrics.Timer;
+import com.google.common.collect.Sets;


Could we use new HashSet<>() from java.util here instead?

kyguy · 2025-10-03T20:34:23Z

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/KafkaCruiseControl.java

                               Long executionProgressCheckIntervalMs,
                               ReplicaMovementStrategy replicaMovementStrategy,
                               Long replicationThrottle,
+                               Long logDirThrottle,


Why not add logDirThrottle to the executeRemoval() method as well? This way we could clear the throttles from the broker metadata when brokers are removed from the cluster.

kyguy · 2025-10-03T20:52:20Z

...se-control/src/test/java/com/linkedin/kafka/cruisecontrol/detector/IdempotenceCacheTest.java

      parameterConfigOverrides.put(MAINTENANCE_EVENT_TYPE_CONFIG, eventType);
      // Expect mocks.
-      EasyMock.expect(_mockKafkaCruiseControl.config()).andReturn(_config).times(4);
+      EasyMock.expect(_mockKafkaCruiseControl.config()).andReturn(_config).anyTimes();


Is there any specific reason we changed this to anyTimes()?

kyguy · 2025-10-03T21:08:01Z

...din/kafka/cruisecontrol/servlet/handler/async/runnable/UpdateTopicConfigurationRunnable.java

          0, _clusterLeaderMovementConcurrency, _brokerLeaderMovementConcurrency,
-          _executionProgressCheckIntervalMs, _replicaMovementStrategy, _replicationThrottle, _isTriggeredByUserRequest, _uuid,
-          SKIP_AUTO_REFRESHING_CONCURRENCY);
+          _executionProgressCheckIntervalMs, _replicaMovementStrategy, _replicationThrottle, null,


The logDirThrottle is null here because partitions are never any intra-broker partition movements as part of topic configuration changes, is my understanding correct?

kyguy · 2025-10-03T21:12:25Z

...java/com/linkedin/kafka/cruisecontrol/servlet/handler/async/runnable/AddBrokersRunnable.java

                                           _executionProgressCheckIntervalMs,
                                           _replicaMovementStrategy,
                                           _replicationThrottle,
+                                           null,


Why is the logDirThrottle argument set to null here? Wouldn't we want new brokers to have the configured logDirThrottle set?

dongjinleekr force-pushed the feature/1851 branch from 4781c13 to c59c864 Compare August 15, 2024 07:44

cpaika approved these changes Aug 15, 2024

View reviewed changes

cpaika mentioned this pull request Sep 13, 2024

Throttling intra broker rebalances #2199

Open

adkafka approved these changes Sep 20, 2024

View reviewed changes

aswinayyolath approved these changes Nov 18, 2024

View reviewed changes

imans777 mentioned this pull request Jan 4, 2025

Rebalance Disks Broker by Broker #2237

Open

CCisGG reviewed Jan 6, 2025

View reviewed changes

dongjinleekr force-pushed the feature/1851 branch from c59c864 to 68b7152 Compare February 23, 2025 14:17

CCisGG requested changes Mar 18, 2025

View reviewed changes

dongjinleekr added 11 commits March 30, 2025 17:09

Remove _throttleRate variable from ReplicationThrottleHelper class

79d77c0

Rename ReplicationThrottleHelper method names: ThrottledRate → Replic…

a6db859

…ationThrottledRate

Rename ReplicationThrottleHelper method name: setThrottles → setRepli…

6879b58

…cationThrottles

Add ReplicationThrottleHelper#LOG_DIR_THROTTLED_RATE support

e5f6489

- Add: setLogDirThrottles, setLogDirThrottledRateIfNecessary - Rename: removeReplicationThrottledRateFromBroker → removeThrottledRatesFromBroker

Add (default.)log.dir.throttle parameter

a6c27cc

Add 'default.log.dir.throttle' to Configurations.md

bea9d53

Remove logDirThrottle parameter in Executor#executeDemoteProposals

6b168bf

Revert method name: removeThrottledRatesFromBroker → removeThrottledR…

76ca5f7

…ateFromBroker

Fix documentation on the 'default.log.dir.throttle' configuration.

411fbbe

Improve Javadoc of KafkaCruiseControl#executeProposals (i.e., inter-b…

685865d

…roker movements vs. intra-broker movements)

dongjinleekr added 2 commits April 2, 2025 23:47

Remove logDirThrottle parameter from KafkaCruiseControl#executeRemoval

fda2ab0

Remove _logDirThrottle from AddBrokersRunnable, RemoveBrokersRunnable…

02057a3

…, and UpdateTopicConfigurationRunnable. (reason: they do not incur any intra-broker replica movements)

dongjinleekr force-pushed the feature/1851 branch from 68b7152 to 02057a3 Compare April 2, 2025 15:46

CCisGG approved these changes Apr 2, 2025

View reviewed changes

This was referenced Apr 7, 2025

[Enhancement]: Support intra-broker throttling on rebalance strimzi/strimzi-kafka-operator#11327

Open

Support intra-broker throttling on rebalance (default.log.dir.throttle) strimzi/strimzi-kafka-operator#11328

Closed

kyguy reviewed Oct 3, 2025

View reviewed changes

Support intra-broker throttling (replica.alter.log.dirs.io.max.bytes.per.second) #2145

Are you sure you want to change the base?

Support intra-broker throttling (replica.alter.log.dirs.io.max.bytes.per.second) #2145

Uh oh!

Conversation

dongjinleekr commented Apr 21, 2024

Uh oh!

dongjinleekr commented Apr 21, 2024

Uh oh!

dongjinleekr commented Jun 20, 2024

Uh oh!

cpaika left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dongjinleekr commented Nov 17, 2024

Uh oh!

adkafka commented Nov 18, 2024

Uh oh!

aswinayyolath commented Nov 18, 2024

Uh oh!

dongjinleekr commented Dec 7, 2024

Uh oh!

CCisGG commented Dec 8, 2024

Uh oh!

imans777 commented Jan 4, 2025

Uh oh!

CCisGG left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

dongjinleekr commented Feb 23, 2025

Uh oh!

luciano-sabenca commented Mar 18, 2025

Uh oh!

CCisGG left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

dongjinleekr commented Apr 2, 2025

Uh oh!

dongjinleekr commented Apr 2, 2025

Uh oh!

CCisGG commented Apr 2, 2025

Uh oh!

CCisGG left a comment

Choose a reason for hiding this comment

Uh oh!

dongjinleekr commented Apr 3, 2025

Uh oh!

kyguy left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment