KAFKA-16944: Rewrite Range Assignor #16504

rreddy-22 · 2024-07-01T23:46:40Z

The server side range assignor was made to be sticky i.e. partitions from the existing assignment are retained as much as possible. During a rebalance, the expected behavior is to achieve co-partitioning for members that are subscribed to the same set of topics with equal number of partitions.

However, there are cases where this cannot be achieved efficiently with the current algorithm. There is no easy way to implement stickiness and co-partitioning and hence we have resorted to recomputing the target assignment every time.

In case of static membership, instanceIds are leveraged to ensure some form of stickiness.

dajac

@rreddy-22 Thanks for the PR. I left some comments for considerations.

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeSet.java

...coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java

rreddy-22 · 2024-07-03T07:30:48Z

TargetAssignmentBuilderBenchmark.build            100                         10           100  avgt    5    0.432 ± 0.003  ms/op
TargetAssignmentBuilderBenchmark.build            100                         10          1000  avgt    5    4.139 ± 0.013  ms/op
TargetAssignmentBuilderBenchmark.build           1000                         10           100  avgt    5    4.332 ± 0.049  ms/op
TargetAssignmentBuilderBenchmark.build           1000                         10          1000  avgt    5   43.449 ± 0.058  ms/op
TargetAssignmentBuilderBenchmark.build          10000                         10           100  avgt    5   47.766 ± 0.389  ms/op
TargetAssignmentBuilderBenchmark.build          10000                         10          1000  avgt    5  487.833 ± 3.459  ms/op

dajac

@rreddy-22 Thanks for the update. I left more comments for consideration. I'd like also point out that they are related failed tests. Could you please check them out?

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeSet.java

...coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeSetTest.java

dajac · 2024-07-03T09:15:15Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeSet.java

+    @Override
+    public Object[] toArray() {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public <T> T[] toArray(T[] a) {
+        throw new UnsupportedOperationException();
+    }


I have noticed that many tests are failing in the build. I suspect that they do because those two methods are not implemented.

From the logs:

java.lang.UnsupportedOperationException at org.apache.kafka.coordinator.group.assignor.RangeSet.toArray(RangeSet.java:83) at java.base/java.util.ArrayList.<init>(ArrayList.java:181) at org.apache.kafka.coordinator.group.CoordinatorRecordHelpers.newTargetAssignmentRecord(CoordinatorRecordHelpers.java:242) at org.apache.kafka.coordinator.group.consumer.TargetAssignmentBuilder.build(TargetAssignmentBuilder.java:368)

The TargetAssignmentBuilderBenchmark results are very likely impacted by this.

…e_Range_Assignor

dajac

LGTM, thanks.

dajac · 2024-07-04T13:42:01Z

Here are the results of the benchmarks based on the last commit:


Benchmark                                       (assignmentType)  (assignorType)  (isRackAware)  (memberCount)  (partitionsToMemberRatio)  (subscriptionType)  (topicCount)  Mode  Cnt    Score    Error  Units
ServerSideAssignorBenchmark.doAssignment             INCREMENTAL           RANGE          false            100                         10         HOMOGENEOUS           100  avgt    5    0.052 ±  0.001  ms/op
ServerSideAssignorBenchmark.doAssignment             INCREMENTAL           RANGE          false            100                         10         HOMOGENEOUS          1000  avgt    5    0.454 ±  0.003  ms/op
ServerSideAssignorBenchmark.doAssignment             INCREMENTAL           RANGE          false           1000                         10         HOMOGENEOUS           100  avgt    5    0.476 ±  0.046  ms/op
ServerSideAssignorBenchmark.doAssignment             INCREMENTAL           RANGE          false           1000                         10         HOMOGENEOUS          1000  avgt    5    3.102 ±  0.055  ms/op
ServerSideAssignorBenchmark.doAssignment             INCREMENTAL           RANGE          false          10000                         10         HOMOGENEOUS           100  avgt    5    5.640 ±  0.223  ms/op
ServerSideAssignorBenchmark.doAssignment             INCREMENTAL           RANGE          false          10000                         10         HOMOGENEOUS          1000  avgt    5   37.947 ±  1.000  ms/op
ServerSideAssignorBenchmark.doAssignment             INCREMENTAL           RANGE          false            100                         10       HETEROGENEOUS           100  avgt    5    0.172 ±  0.001  ms/op
ServerSideAssignorBenchmark.doAssignment             INCREMENTAL           RANGE          false            100                         10       HETEROGENEOUS          1000  avgt    5    1.882 ±  0.006  ms/op
ServerSideAssignorBenchmark.doAssignment             INCREMENTAL           RANGE          false           1000                         10       HETEROGENEOUS           100  avgt    5    1.730 ±  0.036  ms/op
ServerSideAssignorBenchmark.doAssignment             INCREMENTAL           RANGE          false           1000                         10       HETEROGENEOUS          1000  avgt    5   17.654 ±  1.160  ms/op
ServerSideAssignorBenchmark.doAssignment             INCREMENTAL           RANGE          false          10000                         10       HETEROGENEOUS           100  avgt    5   18.595 ±  0.316  ms/op
ServerSideAssignorBenchmark.doAssignment             INCREMENTAL           RANGE          false          10000                         10       HETEROGENEOUS          1000  avgt    5  172.398 ±  2.251  ms/op
JMH benchmarks done

Benchmark                                     (memberCount)  (partitionsToMemberRatio)  (topicCount)  Mode  Cnt   Score   Error  Units
TargetAssignmentBuilderBenchmark.build                  100                         10           100  avgt    5   0.071 ± 0.004  ms/op
TargetAssignmentBuilderBenchmark.build                  100                         10          1000  avgt    5   0.428 ± 0.026  ms/op
TargetAssignmentBuilderBenchmark.build                 1000                         10           100  avgt    5   0.659 ± 0.028  ms/op
TargetAssignmentBuilderBenchmark.build                 1000                         10          1000  avgt    5   3.346 ± 0.102  ms/op
TargetAssignmentBuilderBenchmark.build                10000                         10           100  avgt    5   8.947 ± 0.386  ms/op
TargetAssignmentBuilderBenchmark.build                10000                         10          1000  avgt    5  40.240 ± 3.113  ms/op
JMH benchmarks done

C0urante · 2024-07-04T19:33:15Z

@dajac @rreddy-22 It looks like this commit broke the build since there were new tests added here a few hours before this PR was merged that were not updated to use the new MemberSubscriptionAndAssignmentImpl constructor.

Can we either revert this commit or publish a fix PR ASAP?

C0urante · 2024-07-04T19:43:29Z

Looks like a fix has been published: #16526

tedyu · 2024-07-05T15:55:05Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java

+        GroupSpec groupSpec,
+        SubscribedTopicDescriber subscribedTopicDescriber
+    ) throws PartitionAssignorException {
+        List<String> memberIds = sortMemberIds(groupSpec);


It seems this sorting can be delayed. On line 159, there may be exception thrown.
the sorting can be done when the loop starting on line 157 finishes.

The server side range assignor was made to be sticky i.e. partitions from the existing assignment are retained as much as possible. During a rebalance, the expected behavior is to achieve co-partitioning for members that are subscribed to the same set of topics with equal number of partitions. However, there are cases where this cannot be achieved efficiently with the current algorithm. There is no easy way to implement stickiness and co-partitioning and hence we have resorted to recomputing the target assignment every time. In case of static membership, instanceIds are leveraged to ensure some form of stickiness. ``` Benchmark (assignmentType) (assignorType) (isRackAware) (memberCount) (partitionsToMemberRatio) (subscriptionType) (topicCount) Mode Cnt Score Error Units ServerSideAssignorBenchmark.doAssignment INCREMENTAL RANGE false 100 10 HOMOGENEOUS 100 avgt 5 0.052 ± 0.001 ms/op ServerSideAssignorBenchmark.doAssignment INCREMENTAL RANGE false 100 10 HOMOGENEOUS 1000 avgt 5 0.454 ± 0.003 ms/op ServerSideAssignorBenchmark.doAssignment INCREMENTAL RANGE false 1000 10 HOMOGENEOUS 100 avgt 5 0.476 ± 0.046 ms/op ServerSideAssignorBenchmark.doAssignment INCREMENTAL RANGE false 1000 10 HOMOGENEOUS 1000 avgt 5 3.102 ± 0.055 ms/op ServerSideAssignorBenchmark.doAssignment INCREMENTAL RANGE false 10000 10 HOMOGENEOUS 100 avgt 5 5.640 ± 0.223 ms/op ServerSideAssignorBenchmark.doAssignment INCREMENTAL RANGE false 10000 10 HOMOGENEOUS 1000 avgt 5 37.947 ± 1.000 ms/op ServerSideAssignorBenchmark.doAssignment INCREMENTAL RANGE false 100 10 HETEROGENEOUS 100 avgt 5 0.172 ± 0.001 ms/op ServerSideAssignorBenchmark.doAssignment INCREMENTAL RANGE false 100 10 HETEROGENEOUS 1000 avgt 5 1.882 ± 0.006 ms/op ServerSideAssignorBenchmark.doAssignment INCREMENTAL RANGE false 1000 10 HETEROGENEOUS 100 avgt 5 1.730 ± 0.036 ms/op ServerSideAssignorBenchmark.doAssignment INCREMENTAL RANGE false 1000 10 HETEROGENEOUS 1000 avgt 5 17.654 ± 1.160 ms/op ServerSideAssignorBenchmark.doAssignment INCREMENTAL RANGE false 10000 10 HETEROGENEOUS 100 avgt 5 18.595 ± 0.316 ms/op ServerSideAssignorBenchmark.doAssignment INCREMENTAL RANGE false 10000 10 HETEROGENEOUS 1000 avgt 5 172.398 ± 2.251 ms/op JMH benchmarks done Benchmark (memberCount) (partitionsToMemberRatio) (topicCount) Mode Cnt Score Error Units TargetAssignmentBuilderBenchmark.build 100 10 100 avgt 5 0.071 ± 0.004 ms/op TargetAssignmentBuilderBenchmark.build 100 10 1000 avgt 5 0.428 ± 0.026 ms/op TargetAssignmentBuilderBenchmark.build 1000 10 100 avgt 5 0.659 ± 0.028 ms/op TargetAssignmentBuilderBenchmark.build 1000 10 1000 avgt 5 3.346 ± 0.102 ms/op TargetAssignmentBuilderBenchmark.build 10000 10 100 avgt 5 8.947 ± 0.386 ms/op TargetAssignmentBuilderBenchmark.build 10000 10 1000 avgt 5 40.240 ± 3.113 ms/op JMH benchmarks done ``` Reviewers: David Jacot <[email protected]>

rreddy-22 added 3 commits June 30, 2024 00:01

New Range Assignor

9e553aa

Complete new range assignor - all unit tests pass

e5cd97e

minor

7e772a7

rreddy-22 marked this pull request as ready for review July 1, 2024 23:47

rreddy-22 added 2 commits July 1, 2024 22:54

housekeeping

dc2001e

minor

9fcaf4d

dajac added the KIP-848 The Next Generation of the Consumer Rebalance Protocol label Jul 2, 2024

dajac reviewed Jul 2, 2024

View reviewed changes

address comments

aefd4e5

dajac reviewed Jul 3, 2024

View reviewed changes

rreddy-22 and others added 3 commits July 4, 2024 01:05

address comments

0e7f898

Merge remote-tracking branch 'upstream/trunk' into KAFKA-16944_Rewrit…

2d0f80a

…e_Range_Assignor

fix build

c45ac78

dajac approved these changes Jul 4, 2024

View reviewed changes

dajac merged commit 42f267a into apache:trunk Jul 4, 2024
1 check failed

C0urante mentioned this pull request Jul 4, 2024

KAFKA-16741: Add share group classes for Heartbeat API (1/N) (KIP-932) #16516

Merged

3 tasks

tedyu reviewed Jul 5, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-16944: Rewrite Range Assignor #16504

KAFKA-16944: Rewrite Range Assignor #16504

rreddy-22 commented Jul 1, 2024 •

edited

Loading

dajac left a comment

rreddy-22 commented Jul 3, 2024

dajac left a comment

dajac Jul 3, 2024

dajac Jul 3, 2024

dajac Jul 3, 2024

dajac left a comment

dajac commented Jul 4, 2024

C0urante commented Jul 4, 2024 •

edited

Loading

C0urante commented Jul 4, 2024 •

edited

Loading

tedyu Jul 5, 2024

KAFKA-16944: Rewrite Range Assignor #16504

KAFKA-16944: Rewrite Range Assignor #16504

Conversation

rreddy-22 commented Jul 1, 2024 • edited Loading

dajac left a comment

Choose a reason for hiding this comment

rreddy-22 commented Jul 3, 2024

dajac left a comment

Choose a reason for hiding this comment

dajac Jul 3, 2024

Choose a reason for hiding this comment

dajac Jul 3, 2024

Choose a reason for hiding this comment

dajac Jul 3, 2024

Choose a reason for hiding this comment

dajac left a comment

Choose a reason for hiding this comment

dajac commented Jul 4, 2024

C0urante commented Jul 4, 2024 • edited Loading

C0urante commented Jul 4, 2024 • edited Loading

tedyu Jul 5, 2024

Choose a reason for hiding this comment

rreddy-22 commented Jul 1, 2024 •

edited

Loading

C0urante commented Jul 4, 2024 •

edited

Loading

C0urante commented Jul 4, 2024 •

edited

Loading