Optimize replication throttling in Executor to reduce Kafka admin requests #2214
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…uests Optimizes replication throttling to enhance efficiency during rebalance by reducing redundant Kafka admin requests, thereby improving overall task execution time. Signed-off-by: Aswin A <aswin6303@gmail.com>
CCisGG
requested changes
Oct 29, 2024
| @@ -1644,7 +1656,11 @@ private void interBrokerMoveReplicas() throws InterruptedException, ExecutionExc | |||
| .collect(Collectors.toList()); | |||
| inProgressTasks.addAll(inExecutionTasks()); | |||
|
|
|||
| throttleHelper.clearThrottles(completedTasks, inProgressTasks); | |||
| // Clear throttles only after all tasks are complete for the brokers involved | |||
| if (partitionsToMove == 0) { | |||
There was a problem hiding this comment.
I think you may want to check all of these conditions:
(partitionsToMove == 0 && inExecutionTasks().isEmpty()) || (_stopSignal.get() != NO_STOP_EXECUTION)
| throttleHelper.setThrottles(tasksToExecute.stream().map(ExecutionTask::proposal).collect(Collectors.toList())); | ||
| // Set throttles only if not already set for brokers involved | ||
| List<ExecutionProposal> proposals = tasksToExecute.stream().map(ExecutionTask::proposal).collect(Collectors.toList()); | ||
| for (ExecutionProposal proposal : proposals) { |
There was a problem hiding this comment.
Does it makes more sense to add a map inside throttle helper, so that you don't need to expose the logic in executor. I think the thorttle helper should be responsible to remember which broker is already throttled and just don't send redundant request to those brokers. Same with the clear throttle logic.
There was a problem hiding this comment.
I will take a look
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary
Why:
The current implementation of CC sets and clears replication throttles for each execution task individually, resulting in a high volume of admin requests and slowing down the rebalancing process. This change optimizes throttling by only setting and clearing throttles once per broker during the start and end of a group of tasks, improving efficiency.
What:
Updated
Executor.javato introduce logic for setting throttles per broker at the beginning of task execution and clearing throttles only after all tasks for a broker are complete.Expected Behavior
With this improvement, the replication throttles are set once per broker and cleared after all related tasks are finished, reducing admin request load and improving rebalance speed.
Actual Behavior
Currently, throttling is set and cleared for every individual task, which can cause unnecessary delays during rebalancing due to excessive Kafka admin requests.
Additional Evidence
I have ran the test to make sure all the tests pass.
Additional notes
Made changes like
Now the code uses
getOrDefaultfor accessing task states, which avoids potential NullPointerExceptions if any state is not present in the map.Categorization
This PR resolves
This PR resolves #1972