Port TopicLeaderDistributionGoal and ExecutionTaskPlanner changes - #10
cnmperalta wants to merge 1 commit into
Conversation
Port Liftoff changes made to 2.4.20-liftoff4, originally written by Andrey. This includes: - Replica movement scheduling improvements introduced in #4. - Addition of the TopicLeaderDistributionGoal (#5, #6, c509c47). Some changes in the 3.0.4 version: - Replace leadersOfTopicInBroker and numLeadersOfTopicInBroker with leadersFor and the new built-in numLeadersFor, which both leverage the replicasOfTopicInBroker, which already handles the null case for _topicReplicas.get(topic). - ClusterModel.numTopicLeaders(topic) now directly calls broker.numLeadersFor(topic) for each broker in the cluster instead of going through racks. - Updated ClusterModelStats.numForAvgTopicLeaders to: - Take brokers, topics, and aliveBrokers as parameters to avoid re-computing the same in the method. - Merge the variance computation into the brokers loop, gated on broker.isAlive(), instead of iterating through aliveBrokers separately. - Uses the double primitive instead of Double for BalancingConstraint.topicLeaderBalancePercentage, matching the refactor in the other methods of the class. - Made some style changes so that checkstyle passes. - Added test changes to ExecutionTaskPlannerTest, to reflect the change we made in ExecutionTaskPlanner.
24ceb2b to
1a8e077
Compare
cnmperalta
left a comment
cnmperalta
left a comment
There was a problem hiding this comment.
I marked the areas with code changes I made. The comments are similar to what's in the PR description, just written nearer to the code.
| /** | ||
| * Get leader replicas for topic. | ||
| * @param topic Topic of the requested replicas. | ||
| * @return Leader replicas in this broker sharing the given topic. | ||
| */ | ||
| public Collection<Replica> leadersFor(String topic) { | ||
| return replicasOfTopicInBroker(topic).stream().filter(Replica::isLeader).collect(Collectors.toList()); | ||
| } | ||
|
|
There was a problem hiding this comment.
I replaced leadersOfTopicInBroker and numLeadersOfTopicInBroker with leadersFor and just used the new built-in numLeadersFor, which both leverage the replicasOfTopicInBroker, which already handles the null case for _topicReplicas.get(topic).
| public int numTopicLeaders(String topic) { | ||
| int numTopicLeaders = 0; | ||
| for (Broker broker : brokers()) { | ||
| numTopicLeaders += broker.numLeadersFor(topic); | ||
| } | ||
| return numTopicLeaders; | ||
| } |
There was a problem hiding this comment.
ClusterModel.numTopicLeaders(topic) now directly calls broker.numLeadersFor(topic) for each broker in the cluster instead of going through racks.
| private void numForAvgTopicLeaders(ClusterModel clusterModel, | ||
| SortedSet<Broker> brokers, | ||
| Set<String> topics, | ||
| Set<Broker> aliveBrokers) { |
There was a problem hiding this comment.
Updated ClusterModelStats.numForAvgTopicLeaders to:
- Take
brokers,topics, andaliveBrokersas parameters to avoid re-computing the same in the method. - Merge the
variancecomputation into thebrokersloop, gated onbroker.isAlive(), instead of iterating throughaliveBrokersseparately.
| /** | ||
| * @return Topic leader replica balance percentage for {@link com.linkedin.kafka.cruisecontrol.analyzer.goals.TopicLeaderDistributionGoal}. | ||
| */ | ||
| public double topicLeaderBalancePercentage() { |
There was a problem hiding this comment.
Uses the double primitive instead of Double for BalancingConstraint.topicLeaderBalancePercentage, matching the refactor in the other methods of the class.
| assertEquals("First task", _partitionMovement0, partitionMovementTasks.get(0).proposal()); | ||
| assertEquals("Second task", _partitionMovement2, partitionMovementTasks.get(1).proposal()); | ||
| assertEquals("Third task", _partitionMovement3, partitionMovementTasks.get(2).proposal()); | ||
| assertEquals("Second task", _partitionMovement3, partitionMovementTasks.get(1).proposal()); |
There was a problem hiding this comment.
Added these test changes to ExecutionTaskPlannerTest, to reflect the change we made in ExecutionTaskPlanner.
|
Congrats on lucky PR number 10! 💯 |
|
(Did you mean to mark this as a draft PR?) |
I forgot to mark it as ready for review after I marked it all up. I've done so now haha. |
|
I'm going to be occupied with OCI work for a bit. @andrey-klochkov-liftoff, I don't suppose you have bandwidth to look at this? |
No, sorry, I have enough work, and it's been many years since I saw this code so it'd be an expensive context switch. |
|
Got it. I probably won't be available for this review for a few weeks, unfortunately; up to you if you want to look for another reviewer. |
3 participants
Related to INF-9062.
Port Liftoff changes made to (
2.4.20-liftoff4), originally written by Andrey. This includes:Some changes in the
3.0.4version:leadersOfTopicInBrokerandnumLeadersOfTopicInBrokerwithleadersForand the new built-innumLeadersFor, which both leverage thereplicasOfTopicInBroker, which already handles thenullcase for_topicReplicas.get(topic).ClusterModel.numTopicLeaders(topic)now directly callsbroker.numLeadersFor(topic)for each broker in the cluster instead of going throughracks.ClusterModelStats.numForAvgTopicLeadersto:brokers,topics, andaliveBrokersas parameters to avoid re-computing the same in the method.variancecomputation into thebrokersloop, gated onbroker.isAlive(), instead of iterating throughaliveBrokersseparately.doubleprimitive instead ofDoubleforBalancingConstraint.topicLeaderBalancePercentage, matching the refactor in the other methods of the class.checkstylepasses.ExecutionTaskPlannerTest, to reflect the change we made inExecutionTaskPlanner.Deployment:
3.0.4-liftoff.3.0.4-liftoff1:git tag 3.0.4-liftoff1 3.0.4-liftoff -m 'Port TopicLeaderDistributionGoal and ExecutionTaskPlanner changes to CC 3.0.4'