KAFKA-8179: Part 2, ConsumerCoordinator Algorithm #6778

guozhangwang · 2019-05-20T23:36:34Z

In ConsumerCoordinator, select the protocol as the common protocol from all configured assignor instances' supported protocols with the highest number.

1.b. In onJoinPrepare: only call onPartitionRevoked with EAGER.
1.a. In onJoinComplete: call onPartitionAssigned with EAGER; call onPartitionRevoked following onPartitionAssigned with COOPERATIVE, and then request re-join if the error indicates so.
1.c. In performAssignment: update the user's assignor returned assignments by excluding all partitions that are still owned by some other members.

I've refactored the Subscription / Assignment such that: assigned partitions, error codes, and group instance id are not-final anymore, instead they can be updated. For the last one, it is directly related to the logic of this PR but I felt it is more convienent to go with other fields.
Testing: primarily in ConsumerCoordinatorTest, make it parameterized with protocol, and add necessary scenarios for COOPERATIVE protocol.

I intentionally omitted the documentation change since there are some behavioral updates that needs to be finalized in later PRs, and hence I will also only add the docs in later PRs.

Committer Checklist (excluded from commit message)

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

…-client-algorithm

…pache#6511) As described in KIP-443 (https://cwiki.apache.org/confluence/display/KAFKA/KIP-443%3A+Return+to+default+segment.ms+and+segment.index.bytes+in+Streams+repartition+topics). We want to remove the aggressive overrides of segment.ms and segment.index.bytes for repartition topics. The remaining segment.bytes should still be effective in bounding its footprint. Reviewers: Bill Bejeck <bbejeck@gmail.com>

…-consumer-protocol

…-client-algorithm

guozhangwang · 2019-05-21T21:05:53Z

retest this please

abbccdda · 2019-05-26T20:57:37Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

+        //   2. if there are multiple protocols that are commonly supported, select the one with the highest id (i.e. the
+        //      id number indicates how advanced is the protocol).
+        // we know there are at least one assignor in the list, no need to double check for NPE
+        List<RebalanceProtocol> supportedProtocols = new ArrayList<>(assignors.get(0).supportedProtocols());


Shall we use java doc style comment /** here?

In AK we usually only use /** for function head javadoc.

abbccdda · 2019-05-26T20:58:28Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

+        //   1. only consider protocols that are supported by all the assignors. If there is no common protocols supported
+        //      across all the assignors, throw an exception.
+        //   2. if there are multiple protocols that are commonly supported, select the one with the highest id (i.e. the
+        //      id number indicates how advanced is the protocol).


s/is the protocol/the protocol is

abbccdda · 2019-05-26T21:02:52Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

+
+        switch (protocol) {
+            case EAGER:
+                if (!ownedPartitions.isEmpty()) {


Is this a possible state?

This is not possible state, and hence I'm logging a WARN. I can also just throw an exception actually.

abbccdda · 2019-05-26T21:06:26Z

clients/src/main/java/org/apache/kafka/common/utils/Bytes.java

@@ -141,7 +141,7 @@ private static String toString(final byte[] b, int off, int len) {
    }

    /**
-     * A byte array comparator based on lexicograpic ordering.
+     * A byte array comparator based on lexicographic ordering.


nice typo fix

abbccdda · 2019-05-26T21:08:59Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

+        }
+
+        if (supportedProtocols.isEmpty()) {
+            throw new IllegalArgumentException("Specified assignors do not have commonly supported rebalance protocol");


Do we want to include assignor value here?

Good point.

abbccdda · 2019-05-26T21:16:01Z

clients/src/test/java/org/apache/kafka/test/TestUtils.java

@@ -434,6 +435,10 @@ public static void isValidClusterId(String clusterId) {
        return list;
    }

+    public static <T> Set<T> toSet(Collection<T> collection) {


Looks like we could use this for a lot of refactoring :)

…-consumer-protocol

ableegoldman · 2019-06-13T05:29:35Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

-        log.info("Setting newly assigned partitions: {}", Utils.join(assignedPartitions, ", "));
+
+        switch (protocol) {
+            case EAGER:


Slight readability improvement: could we factor this into a method, maybe #assignPartitions? We could also then have just #revokePartitions and the cooperative case could call both -- this way it's immediately clear what each protocol does/how they differ

I tried to update it in the way you suggested, but it turns out a bit tricky since for some cases we do want to update the owned-partitions state if COOPERATIVE is used, while for some others we do not rely on the protocol to decide if the owned partitions would be revoked or not..

I agree you'd have to factor out the logic of what exactly is being revoked/assigned into the switch statement and then call assign/revoke on the appropriate set of partitions. Might be worth doing if we add another protocol but for now it's not that important

guozhangwang · 2019-06-18T22:00:02Z

@ableegoldman @mjsax @hachikuji @abbccdda ready for another review. Also updating the description on the top to summarize the change here.

ableegoldman · 2019-06-18T22:29:58Z

clients/src/test/java/org/apache/kafka/clients/MockClient.java

@@ -221,7 +221,7 @@ public void send(ClientRequest request, long now) {
                    builder.latestAllowedVersion());
            AbstractRequest abstractRequest = request.requestBuilder().build(version);
            if (!futureResp.requestMatcher.matches(abstractRequest))
-                throw new IllegalStateException("Request matcher did not match next-in-line request " + abstractRequest);
+                throw new IllegalStateException("Request matcher did not match next-in-line request " + abstractRequest + " with prepared response " + futureResp.responseBody);


nit: break up long line

@ableegoldman Sorry I missed these comments. Will address in the pt.3 PR.

ableegoldman

Few nits, but overall Part 2 LGTM

ableegoldman · 2019-06-18T22:45:38Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java

@@ -176,6 +196,25 @@ public void teardown() {
        this.coordinator.close(time.timer(0));
    }

+    @Test
+    public void testSelectRebalanceProtcol() {


Suggested change

public void testSelectRebalanceProtcol() {

public void testSelectRebalanceProtocol() {

ableegoldman · 2019-06-18T22:57:41Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java

    }

    @Test
    public void testMetadataRefreshDuringRebalance() {
        final String consumerId = "leader";
+        final List<TopicPartition> owned = Collections.emptyList();
+        final List<TopicPartition> oldAssigned = Arrays.asList(t1p);
+


nit: extra line

ableegoldman · 2019-06-18T23:00:43Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java

-        coordinator.poll(time.timer(Long.MAX_VALUE));
-        assertFalse(client.hasPendingResponses());
+        try (ConsumerCoordinator coordinator = buildCoordinator(rebalanceConfig, new Metrics(), assignors,
+            true)


nit: all on same line (or all separate)?

ableegoldman · 2019-06-18T23:01:19Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java

-
-        subscriptions.subscribe(singleton(topic1), rebalanceListener);
+        try (ConsumerCoordinator coordinator = buildCoordinator(rebalanceConfig, new Metrics(), assignors,
+            true)) {


guozhangwang · 2019-06-18T23:17:31Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java

@@ -2294,7 +2399,7 @@ private JoinGroupResponse joinGroupLeaderResponse(int generationId,
                        .setProtocolName(partitionAssignor.name())
                        .setLeader(memberId)
                        .setMemberId(memberId)
-                        .setMembers(Collections.emptyList())
+                        .setMembers(metadata)


This is a piggy-backed fix: we should encode the metadata for leaders.

bbejeck

Just a couple of minor nits but LGTM.

bbejeck · 2019-06-21T16:09:46Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

+        switch (protocol) {
+            case EAGER:
+                if (!ownedPartitions.isEmpty()) {
+                    log.info("Coordinator has owned partitions {} that are not revoked with {} protocol, " +


Did this slip? Seems it should be a WARN as you said above.

I originally make it an exception since it should not happen, but later in unit test I realized it could normally happen if we have a wakeup exception during rebalance listener call, and therefore I change it to INFO

bbejeck · 2019-06-21T17:22:48Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocol.java


            // assume all higher versions can be parsed as V1
            default:
-                return buildSubscriptionV1(buffer, groupInstanceId);
+                return deserializeSubscriptionV1(buffer);


nit: Although not strictly part of the PR, this part isn't covered in the unit test. Is it worth mocking out future version /different protocol to make sure this works as intended? Same for serializeSubscription

Could you elaborate a bit more? I thought future-proof is covered in deserializeFutureSubscription/AssignmentVersion`.

My bad, I missed the tests deserializeFutureSubscription/AssignmentVersion and was looking at the coverage of de/serializeAssignment, and the default branch isn't hit, but the individual methods themselves are covered.

hachikuji · 2019-06-24T23:14:50Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

+
+            protocol = supportedProtocols.get(supportedProtocols.size() - 1);
+        } else {
+            protocol = null;


Won't this break in the switch below? Am I missing something?

We have a check at the beginning of ConsumerCoordinator.poll:

if (protocol == null) { throw new IllegalStateException("User confingure ConsumerConfig#PARTITION_ASSIGNMENT_STRATEGY_CONFIG to empty " + "while trying to subscribe for group protocol to auto assign partitions"); }

Basically if user does not call subscribe() it is okay to not set the protocol at all; otherwise we will throw error.

Ok. We also have a check in subscribe to ensure that the set of assignors is not empty. There might be a way to remove the redundant checking.

By the way, there's a typo above: confingure.

Yeah I think we can indeed remove it:

if no assignors specified, then subscribe will throw.

if assignors specified but no common supported protocol, then constructor will throw.

hachikuji · 2019-06-24T23:26:45Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

-        Map<String, Assignment> assignment = assignor.assign(metadata.fetch(), subscriptions);
+        Map<String, Assignment> assignments = assignor.assign(metadata.fetch(), subscriptions);
+
+        switch (protocol) {


nit: since we're not doing anything in the EAGER case, couldn't we simplify this:

if (protocol == COOPERATIVE) adjustAssignment(ownedPartitions, assignments)

Similarly in onJoinPrepare

Ack here; For onJoinPrepare though in the pt.3 this is modified a bit since with EAGER we may still revoke.

hachikuji · 2019-06-24T23:59:12Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

+            case EAGER:
+                if (!ownedPartitions.isEmpty()) {
+                    log.info("Coordinator has owned partitions {} that are not revoked with {} protocol, " +
+                        "it is likely client is woken up before a previous pending rebalance completes its callback", ownedPartitions, protocol);


I think this log message is more confusing than useful. Maybe it would be better to log a message below in the Wakeup/Interrupt exception handler?

hachikuji · 2019-06-25T00:09:04Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

+
+    }
+
+    private void assignAndRevoke(final ConsumerRebalanceListener listener,


It would be helpful to have some high-level documentation somewhere in the code about the cooperative strategy and how it works. Perhaps this already exists? At a minimum, we should explain in the ConsumerRebalanceListener docs how the two strategies affect the API.

Yeah my plan is to add the javadoc change in pt.3, and will do another doc PR for web-docs / upgrade-path; the rationale is that some of the behavior changes are yet to be agreed on in pt.3 so I'd like to defer that to later.

hachikuji

Left a few comments.

hachikuji · 2019-06-25T00:13:35Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

+                break;
+
+            case COOPERATIVE:
+                adjustAssignment(ownedPartitions, assignments);


A brief description of the purpose of this adjustment would be helpful.

hachikuji · 2019-06-25T00:18:45Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PartitionAssignor.java

@@ -186,6 +182,10 @@ public ByteBuffer userData() {
            return userData;
        }

+        public void setGroupInstanceId(Optional<String> groupInstanceId) {


Turning a final field into a mutable field seems like a step backwards... Can you explain why this is needed?

hachikuji · 2019-06-25T00:29:00Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PartitionAssignor.java

     * By default it should always work with {@link RebalanceProtocol#EAGER}.
     */
    default List<RebalanceProtocol> supportedProtocols() {
        return Collections.singletonList(RebalanceProtocol.EAGER);
    }

    /**
-     * Return the version of the assignor which indicate how the user metadata encodings
+     * Return the version of the assignor which indicates how the user metadata encodings


I think I mentioned this before as well, but can we elaborate on the use of this version? This is very vague.

The version of the assignor would be used at the broker-side to select the leader as the one with the highest version, expecting it be able to decode assignor's older subscription user metadata.

I will update the javadoc as well after the broker's logic has been updated accordingly.

hachikuji · 2019-06-25T00:29:22Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PartitionAssignor.java

            return error;
        }

+        public void updatePartitions(List<TopicPartition> partitions) {


Do these methods have to be public?

Ah good catch, it needs to be default-access.

hachikuji · 2019-06-25T00:39:46Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

            ByteBuffer buffer = ConsumerProtocol.serializeAssignment(assignmentEntry.getValue());
            groupAssignment.put(assignmentEntry.getKey(), buffer);
        }

        return groupAssignment;
    }

+    private void adjustAssignment(final Map<TopicPartition, String> ownedPartitions,


We may have discussed this in the past, but I'm not too thrilled with the consumer mucking around with assignments from the plugin. Couldn't we just raise an error if the assignment doesn't respect ownership?

Couldn't we just raise an error if the assignment doesn't respect ownership?

Hmm, I think that's doable. But on the other hand we are requiring users to be aware how to leverage on owned-partitions list and respect it with COOPERATIVE.

I see pros and cons of this approach: not silently mucking around user assignor returned data, but pushing the responsibility of implementing COOPERATIVE-supported assignors to users themselves. @ableegoldman wdyt?

hachikuji · 2019-06-25T00:43:23Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PartitionAssignor.java

        private final ByteBuffer userData;
-        private final ConsumerProtocol.Errors error;
+        private ConsumerProtocol.AssignmentError error;


Does the assignor need to be aware of this error? If it is just for internal use, can we leave it out of the public API?

guozhangwang added 30 commits April 1, 2019 16:20

bump up consumer protocol to v2

4cf89e6

update consumer coordinator based on rebalance protocol

094ab43

Merge branch 'trunk' of https://github.com/apache/kafka into K8179-p2…

368f2bc

…-client-algorithm

minor fix

cb9fb49

set re-join error code

f7ab5e2

github comments

cb25f6c

rebase from p1

801dd0b

remove logic for handling rebalance callback errors

0a4aa1d

fixes on unit tests, add more unit tests

3f2fe2e

Merge branch 'trunk' of https://github.com/apache/kafka into K8179-p1…

bb98793

…-consumer-protocol

add version varible to subscription and assignment

d1064c4

rebase from p1

829d3c2

more minor fixes

2b2d8b1

update toString

c22266b

Merge branch 'K8179-p1-consumer-protocol' into K8179-p2-client-algorithm

1fa089a

minor fix

16f75ab

found another lurking bug on my code

e049e5f

rebase from trunk

92320ba

new API

ff1b9a6

add version API

908fda4

Merge branch 'trunk' of https://github.com/apache/kafka into K8179-p1…

3e1fb1c

…-consumer-protocol

github comments

2e9a291

Merge branch 'trunk' of https://github.com/apache/kafka into K8179-p1…

bd60fd2

…-consumer-protocol

Merge branch 'trunk' of https://github.com/apache/kafka into K8179-p1…

74a9895

…-consumer-protocol

rebase from trunk

c85ed6e

throw from user rebalance callback

16c13ad

fix unit tests

ec80f42

Merge branch 'trunk' of https://github.com/apache/kafka into K8179-p2…

60b55e1

…-client-algorithm

fix unit test ConsumerCoordinatorTest

5325e56

add more unit tests

72e70d6

guozhangwang changed the title ~~KAFKA-8179: Part 2, ConsumerCoordinator Algorithm [WIP]~~ KAFKA-8179: Part 2, ConsumerCoordinator Algorithm May 21, 2019

fix checkstyle

ba97f99

abbccdda reviewed May 26, 2019

View reviewed changes

guozhangwang added 3 commits June 4, 2019 17:54

Merge branch 'trunk' of https://github.com/apache/kafka into K8179-p1…

e54d8e1

…-consumer-protocol

rebase from trunk

dfb3583

github comments

99cafcd

ableegoldman reviewed Jun 13, 2019

View reviewed changes

guozhangwang added 4 commits June 18, 2019 13:28

rebase from trunk

dd37aa1

update tests

cf66222

minor fixes

d56b1f3

github comments

f67942f

ableegoldman reviewed Jun 18, 2019

View reviewed changes

ableegoldman approved these changes Jun 18, 2019

View reviewed changes

guozhangwang commented Jun 18, 2019

View reviewed changes

guozhangwang added 2 commits June 18, 2019 17:24

set protocol with assignors configured only

11b1915

maintain consistent behavior

8440e7d

bbejeck approved these changes Jun 21, 2019

View reviewed changes

guozhangwang merged commit e5c4ebd into apache:trunk Jun 23, 2019

hachikuji reviewed Jun 24, 2019

View reviewed changes

hachikuji reviewed Jun 25, 2019

View reviewed changes

This was referenced Jul 23, 2019

KAFKA-8179: KIP-429, new PartitionAssignor interface #7100

Closed

KAFKA-8179: add public ConsumerPartitionAssignor interface #7108

Merged

guozhangwang deleted the K8179-p2-client-algorithm branch April 24, 2020 23:58

	public void testSelectRebalanceProtcol() {
	public void testSelectRebalanceProtocol() {


		}

		private void assignAndRevoke(final ConsumerRebalanceListener listener,

KAFKA-8179: Part 2, ConsumerCoordinator Algorithm #6778

KAFKA-8179: Part 2, ConsumerCoordinator Algorithm #6778

Conversation

guozhangwang commented May 20, 2019 • edited

Committer Checklist (excluded from commit message)

guozhangwang commented May 21, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented Jun 18, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ableegoldman left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji Jun 25, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented May 20, 2019 •

edited

hachikuji Jun 25, 2019 •

edited