KAFKA-6718 / Rack aware standby task assignor by lkokhreidze · Pull Request #10851 · apache/kafka

lkokhreidze · 2021-06-09T10:02:49Z

This PR is part of KIP-708 and adds rack aware standby task assignment logic.

Rack aware standby task assignment won't be functional until all parts of this KIP gets merged.

Splitting PRs into three smaller PRs to make the review process easier to follow. Overall plan is the following:

👉 Rack aware standby task assignment logic.
⏭️ Protocol change, add clientTags to SubscriptionInfoData #10802
⏭️ Add required configurations to StreamsConfig (public API change, at this point we should have full functionality) #11837

This PR implements first point of the above mentioned plan.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

lkokhreidze · 2021-06-09T10:04:17Z

Call for review @cadonna @vvcephei @ableegoldman

lkokhreidze · 2021-06-16T08:33:17Z

gentle nudge @cadonna @vvcephei @ableegoldman

cadonna · 2021-07-13T12:08:48Z

@lkokhreidze Sorry for the long silence! Some other tasks got in my way. I plan to review this PR tomorrow.

cadonna

@lkokhreidze Thank you for the PR!

I did a first pass. Here is my feedback.

…nals/assignment/DefaultStandbyTaskAssignor.java Co-authored-by: Bruno Cadonna <cadonna@apache.org>

…; Remove NoopStandbyTaskAssignor;

lkokhreidze · 2021-07-14T18:46:39Z

Thanks @cadonna for the feedback.
I've replied/addressed all of your comments.

cadonna

@lkokhreidze Thank you for the updates!

Here my next round of feedback!

…ignature change of HighAvailabilityTaskAssignor#assignActiveStatefulTasks

lkokhreidze · 2021-07-19T06:41:50Z

Thanks @cadonna for the valuable feedback 🙇
I've replied/addressed your comments.

cadonna · 2021-07-23T13:38:45Z

@lkokhreidze FYI: I will be offline the next two weeks. I am sorry that I haven't had time to review this PR this week.

cadonna

Hi @lkokhreidze !

Thank you for your patience!

Here my feedback.

lkokhreidze · 2021-08-16T06:01:13Z

Hi @cadonna !
Thanks for the feedback. I will address your comments this week.

…reidze/kafka into KAFKA-6718-standby-task-assignment

lkokhreidze · 2022-02-21T11:08:33Z

Thanks @showuon for the valuable feedback.
I've addressed your comments, please have a look when you got time.

…mberOfStandbyTasksEqualsTo

showuon

@lkokhreidze , thanks for the update. Overall LGTM! Only one question for partial rack awareness assignment. Thanks.

…yOfTheTagEntries

cadonna · 2022-02-22T14:06:40Z

+            // If we have used more clients than all the tag's unique values,
+            // we can't filter out clients located on that tag, because it'll exclude all the clients.
+            // Instead, we need to discard all the clients that were marked as "used" on that tag key values.
+            // Please check ClientTagAwareStandbyTaskAssignorTest#shouldDoThePartialRackAwareness test for more info.


That is quite challenging to understand. After reading it a couple of times I understood that if we've used a number of clients that is equal to or greater than the number of unique values of the tag, we cannot guarantee that each standby is on a different tag value than the active and other standbys. So the rack-awareness becomes partial. Is that correct?
Could you reformulate it, so that it states that the rack-awareness guarantee does not hold anymore.
And why "more clients than all tag's unique values"? When the number of used clients is equal to the unique tag values, we are already in the partial rack-awareness situation, right?
Maybe you should give here an example as in the mentioned test. I find referencing the test is a bit cumbersome, because if the test gets renamed this comment becomes useless.

Thanks Bruno, I'll add the example to the comment.

Pushed the changes. I added detailed explanation with an example. Also tests have the similar example. Hopefully this change makes logic more clear.

cadonna · 2022-02-22T14:41:55Z

+        final int numStandbyReplicas = configs.numStandbyReplicas;
+        final Set<String> rackAwareAssignmentTags = new HashSet<>(configs.rackAwareAssignmentTags);
+
+        final Map<TaskId, Integer> tasksToRemainingStandbys = computeTasksToRemainingStandbys(


I could not find where you decrement the number of remaining standbys. If you get a value from this map and put it into an int variable, you do not have a reference to the Integer value in the map anymore. This might become a problem in StandbyTaskAssignmentUtils#pollClientAndMaybeAssignRemainingStandbyTasks().

Nice catch! And maybe we should add a test to address this.

Yes, a test is absolutely necessary!

Updated tests for the StandbyTaskAssignmentUtils#pollClientAndMaybeAssignAndUpdateRemainingStandbyTasks and also added separate test for the ClientTagAwareStandbyTaskAssignor.

For the ClientTagAwareStandbyTaskAssignor I decided to make few things package private to be able to test this logic. As otherwise, there was no easy way to test if tasksToRemainingStandbys was getting updated properly. Hope this is okay.

…nals/assignment/ClientTagAwareStandbyTaskAssignor.java Co-authored-by: Bruno Cadonna <cadonna@apache.org>

showuon

Thanks for the update in partial rack awareness assignment. LGTM! Left some minor comments. Thanks.

showuon · 2022-02-23T03:09:25Z

+        final int numStandbyReplicas = configs.numStandbyReplicas;
+        final Set<String> rackAwareAssignmentTags = new HashSet<>(configs.rackAwareAssignmentTags);
+
+        final Map<TaskId, Integer> tasksToRemainingStandbys = computeTasksToRemainingStandbys(


Nice catch! And maybe we should add a test to address this.

lkokhreidze · 2022-02-23T09:56:15Z

Hi @cadonna @showuon
I've addressed your comments. Please have a look.
Thank you!

showuon

Thanks for the update. I'll check the test code tomorrow.

showuon

LGTM! Thanks for the improvement!

lkokhreidze · 2022-02-24T16:38:30Z

Hi @cadonna
Mind having another look? Hoping to finalise this PR as soon as possible :)
Thanks

cadonna · 2022-03-01T13:26:38Z

@lkokhreidze I will try to have a look this week. I would also like to get this PR merged as soon as possible. Since @showuon has already approved it. If I will not manage to have a look, @showuon can merge it.

lkokhreidze · 2022-03-01T14:14:22Z

Thanks @cadonna, appreciate it.
@showuon please also note that next PR, the protocol change, is also ready to be reviewed. #10802

Whenever it's possible, please have a look at that too.

cadonna

@lkokhreidze Thanks a lot for this PR!

Sorry for all the delay!

LGTM! I had just some minor comments that can be addressed in a follow-up PR.

As I pointed out in one of my comments, I am not a big fan of // Visible for testing because that often indicates some deficiencies in the modularization. But this is nothing that cannot be addressed in a follow-up PR. So I am fine with merging it.

cadonna · 2022-03-02T12:08:36Z

+        }
+
+        if (!tasksToRemainingStandbys.isEmpty()) {
+            log.debug("Rack aware standby task assignment was not able to assign all standby tasks. " +


I think info log would also be OK here. I imagine users that are wondering why their standbys are not distributed as they would expect. With this information they can at least try to fix it on the config level. This log message should only happen at rebalance time, which should usually be rather seldom.
If we decide to put the log message on info level, you should also change a bit the wording and not use variable names in it. Maybe some hints what the users can do to fix this would also be nice.

Is it possible to separate the concerns of this log message and the one on line 135? Something along the lines of here the rack-aware standby assignment did not work due the tag config and on line 135 the assignment did not work due to too low number of instances. We can then put both on warn or info (do not forget to also check the related log message in the default standby assignor).

cadonna · 2022-03-02T12:37:25Z

+    // Visible for testing
+    static StandbyTaskAssignor createStandbyTaskAssignor(final AssignmentConfigs configs) {
+        if (!configs.rackAwareAssignmentTags.isEmpty()) {
+            return new ClientTagAwareStandbyTaskAssignor();
+        } else {
+            return new DefaultStandbyTaskAssignor();
+        }
+    }
+


This can be done in a follow-up PR:
I am not a big fan of // Visible for testing because it often means that we missed to extract code to separate classes. Here I would definitely extract this code to a factory class.

cadonna · 2022-03-02T13:15:29Z

@lkokhreidze Let me know if you want to address my minor comments in this PR. After that @showuon or I can merge this PR.

Nice work!

…nals/assignment/ClientTagAwareStandbyTaskAssignor.java Co-authored-by: Bruno Cadonna <cadonna@apache.org>

lkokhreidze · 2022-03-02T13:37:19Z

Hi @cadonna , thanks for the review.
I agree with what you said and made a note to myself to address your comments in the follow-up PRs. So if it's okay, I think we can merge this one.

Thank you!

showuon · 2022-03-03T03:30:46Z

Failed tests are unrelated and also failed in trunk build.

Build / ARM / org.apache.kafka.streams.integration.NamedTopologyIntegrationTest.shouldRemoveOneNamedTopologyWhileAnotherContinuesProcessing
Build / JDK 11 and Scala 2.13 / kafka.server.DynamicBrokerReconfigurationTest.testThreadPoolResize()
Build / JDK 11 and Scala 2.13 / kafka.server.DynamicBrokerReconfigurationTest.testThreadPoolResize()
Build / JDK 11 and Scala 2.13 / org.apache.kafka.streams.integration.NamedTopologyIntegrationTest.shouldRemoveOneNamedTopologyWhileAnotherContinuesProcessing
Build / JDK 17 and Scala 2.13 / kafka.server.DynamicBrokerReconfigurationTest.testThreadPoolResize()
Build / JDK 17 and Scala 2.13 / kafka.server.DynamicBrokerReconfigurationTest.testThreadPoolResize()
Build / JDK 17 and Scala 2.13 / org.apache.kafka.streams.integration.NamedTopologyIntegrationTest.shouldRemoveOneNamedTopologyWhileAnotherContinuesProcessing
Build / JDK 8 and Scala 2.12 / org.apache.kafka.connect.integration.RebalanceSourceConnectorsIntegrationTest.testDeleteConnector
Build / JDK 8 and Scala 2.12 / org.apache.kafka.connect.integration.RebalanceSourceConnectorsIntegrationTest.testDeleteConnector
Build / JDK 8 and Scala 2.12 / org.apache.kafka.streams.integration.NamedTopologyIntegrationTest.shouldRemoveOneNamedTopologyWhileAnotherContinuesProcessing

cadonna · 2022-03-03T10:28:49Z

🥳

This PR is part of KIP-708 and adds rack aware standby task assignment logic. Rack aware standby task assignment won't be functional until all parts of this KIP gets merged. Splitting PRs into three smaller PRs to make the review process easier to follow. Overall plan is the following: ⏭️ Rack aware standby task assignment logic #10851 ⏭️ Protocol change, add clientTags to SubscriptionInfoData #10802 👉 Add required configurations to StreamsConfig (public API change, at this point we should have full functionality) This PR implements last point of the above mentioned plan. Reviewers: Luke Chen <showuon@gmail.com>, Bruno Cadonna <cadonna@apache.org>

KAFKA-6718 / Rack aware standby task assignor

6d48ef8

lkokhreidze mentioned this pull request Jun 9, 2021

KAFKA-6718 / Update SubscriptionInfoData with clientTags #10802

Merged

3 tasks

Merge branch 'trunk' into KAFKA-6718-standby-task-assignment

0aa7689

Merge branch 'trunk' into KAFKA-6718-standby-task-assignment

5647a2b

vlsi reviewed Jul 7, 2021

View reviewed changes

Comment thread streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ClientState.java Outdated

vlsi reviewed Jul 7, 2021

View reviewed changes

Comment thread ...c/main/java/org/apache/kafka/streams/processor/internals/assignment/StandbyTaskAssignor.java Outdated

Merge branch 'trunk' into KAFKA-6718-standby-task-assignment

c875ca8

mjsax added the streams label Jul 13, 2021

cadonna reviewed Jul 14, 2021

View reviewed changes

lkokhreidze and others added 3 commits July 14, 2021 16:25

Update streams/src/main/java/org/apache/kafka/streams/processor/inter…

3b4244f

…nals/assignment/DefaultStandbyTaskAssignor.java Co-authored-by: Bruno Cadonna <cadonna@apache.org>

Merge branch 'trunk' into KAFKA-6718-standby-task-assignment

61aa369

Change TreeMap to SortedMap in StandbyTaskAssignor#assignStandbyTasks…

bbf6394

…; Remove NoopStandbyTaskAssignor;

cadonna reviewed Jul 15, 2021

View reviewed changes

KAFKA-6718 / Move factory method in the StandbyTaskAssignor; Revert s…

47cb6b9

…ignature change of HighAvailabilityTaskAssignor#assignActiveStatefulTasks

lkokhreidze added 4 commits July 19, 2021 10:37

KAFKA-6718 / Remove emptry constructor

f0dbb47

KAFKA-6718 / Revert changes in HighAvailabilityTaskAssignor

eecabb2

KAFKA-6718 / Revert changes in HighAvailabilityTaskAssignor

494a18f

KAFKA-6718 / Revert changes in HighAvailabilityTaskAssignor

9dbfb01

KAFKA-6718 / Sync with trunk

a0b5a04

cadonna reviewed Aug 10, 2021

View reviewed changes

lkokhreidze added 3 commits August 20, 2021 20:13

Merge branch 'trunk' into KAFKA-6718-standby-task-assignment

bcb8e94

CR

9841d25

KAFKA-6718 / Fix bug

4505453

lkokhreidze added 2 commits February 21, 2022 12:53

KAFKA-6718 / CR

b164024

Merge branch 'KAFKA-6718-standby-task-assignment' of github.com:lkokh…

cb565a8

…reidze/kafka into KAFKA-6718-standby-task-assignment

KAFKA-6718 / Rename assertAllStandbyTasksAreAssigned -> assertTotalNu…

1b2d8ff

…mberOfStandbyTasksEqualsTo

showuon reviewed Feb 22, 2022

View reviewed changes

lkokhreidze added 2 commits February 22, 2022 14:57

KAFKA-6718 / CR

5d1543e

KAFKA-6718 / Rename isClientInAnyOfTheUsedClients -> isClientUsedOnAn…

06ecb6d

…yOfTheTagEntries

cadonna reviewed Feb 22, 2022

View reviewed changes

Update streams/src/main/java/org/apache/kafka/streams/processor/inter…

cfe6280

…nals/assignment/ClientTagAwareStandbyTaskAssignor.java Co-authored-by: Bruno Cadonna <cadonna@apache.org>

showuon reviewed Feb 23, 2022

View reviewed changes

KAFKA-6718 / CR

040cba0

showuon reviewed Feb 23, 2022

View reviewed changes

Comment thread ...g/apache/kafka/streams/processor/internals/assignment/ClientTagAwareStandbyTaskAssignor.java Outdated

lkokhreidze added 2 commits February 23, 2022 15:33

Update ClientTagAwareStandbyTaskAssignor.java

0398277

Update StandbyTaskAssignmentUtils.java

eaf3039

showuon approved these changes Feb 24, 2022

View reviewed changes

Merge branch 'trunk' into KAFKA-6718-standby-task-assignment

3b4e9f4

cadonna approved these changes Mar 2, 2022

View reviewed changes

lkokhreidze and others added 2 commits March 2, 2022 15:27

Update streams/src/main/java/org/apache/kafka/streams/processor/inter…

9870eee

…nals/assignment/ClientTagAwareStandbyTaskAssignor.java Co-authored-by: Bruno Cadonna <cadonna@apache.org>

Update streams/src/main/java/org/apache/kafka/streams/processor/inter…

3f5c81f

…nals/assignment/ClientTagAwareStandbyTaskAssignor.java Co-authored-by: Bruno Cadonna <cadonna@apache.org>

showuon merged commit 62e6466 into apache:trunk Mar 3, 2022

lkokhreidze mentioned this pull request Mar 3, 2022

KAFKA-6718 / Add rack awareness configurations to StreamsConfig #11837

Merged

3 tasks

Conversation

lkokhreidze commented Jun 9, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

lkokhreidze commented Jun 9, 2021

Uh oh!

lkokhreidze commented Jun 16, 2021

Uh oh!

Uh oh!

Uh oh!

cadonna commented Jul 13, 2021

Uh oh!

cadonna left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

lkokhreidze commented Jul 14, 2021

Uh oh!

cadonna left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

lkokhreidze commented Jul 19, 2021

Uh oh!

cadonna commented Jul 23, 2021

Uh oh!

cadonna left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

lkokhreidze commented Aug 16, 2021

Uh oh!

lkokhreidze commented Feb 21, 2022

Uh oh!

showuon left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

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!

lkokhreidze Feb 23, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

showuon left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lkokhreidze commented Jun 9, 2021 •

edited

Loading

lkokhreidze Feb 23, 2022 •

edited

Loading