KAFKA-13029; Set appropriate fields for FindCoordinatorRequest based on version #10965

rajinisivaram · 2021-07-03T10:33:23Z

KIP-699 added support for batching in FindCoordinatorRequest using a new protocol that changes the wire format for both batched and unbatched requests. Clients were updated to try the new format first and switch irreversibly to the old format if the new format is not supported on one broker. During rolling upgrade (or a downgrade), it is possible that a broker doesn't support new format at some point while other brokers do at a later point. Clients end up in a bad state until restarted since use new version with old format. This PR changes FindCoordinatorRequest to set data based on actual version when a single group is used. This is always the case for consumer coordinator and transaction coordinator. For admin API, we still switch to unbatched mode on failure, but the data is set based on actual version, so we never fail even if brokers are upgraded/downgraded.

Committer Checklist (excluded from commit message)

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

…on version

showuon

@rajinisivaram , thanks for the PR. Nice tests! Left some comments.

showuon · 2021-07-04T07:47:23Z

clients/src/main/java/org/apache/kafka/common/requests/FindCoordinatorRequest.java


 public class FindCoordinatorRequest extends AbstractRequest {

+    public static final short MIN_BATCHED_VERSION = 4;


Nice refactor!

showuon · 2021-07-04T07:51:42Z

clients/src/main/java/org/apache/kafka/common/requests/FindCoordinatorRequest.java

+            if (version < MIN_BATCHED_VERSION) {
+                if (batchedKeys > 1)
+                    throw new NoBatchedFindCoordinatorsException("Cannot create a v" + version + " FindCoordinator request " +
                        "because we require features supported only in 4 or later.");


nit: "because we require features supported only in" + MIN_BATCHED_VERSION + "or later.");

showuon · 2021-07-04T07:55:13Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

        abstract AbstractRequest.Builder<?> requestBuilder();

-        abstract void handleResponse(AbstractResponse responseBody);
+        abstract void handleResponse(AbstractResponse responseBody, short requestVersion);


maybe we can add a comment or java doc here to describe why requestVersion is necessary here.

showuon · 2021-07-04T08:05:06Z

clients/src/main/java/org/apache/kafka/clients/admin/internals/CoordinatorStrategy.java

        }
        FindCoordinatorResponse response = (FindCoordinatorResponse) abstractResponse;
-        if (batch) {
+        if (!response.data().coordinators().isEmpty()) {


could we add a comment here to explain what case it is (ex: batch response case)

rajinisivaram · 2021-07-04T11:19:05Z

@showuon Thanks for the review, addressed comments.

mimaison

Thanks @rajinisivaram for this fix!

chia7712

@rajinisivaram thanks for this nice patch. This patch also resolves the slow poll when requesting to older broker (#10963). a couple of comments below. PTAL

chia7712 · 2021-07-04T13:01:45Z

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

-        } else {
-            data.setKey(this.rebalanceConfig.groupId);
-        }
+        data.setKey(this.rebalanceConfig.groupId);


this line can be merged. for example:

FindCoordinatorRequestData data = new FindCoordinatorRequestData() .setKeyType(CoordinatorType.GROUP.id()) .setKey(this.rebalanceConfig.groupId);

chia7712 · 2021-07-04T13:51:00Z

clients/src/main/java/org/apache/kafka/clients/admin/internals/CoordinatorStrategy.java

        FindCoordinatorResponse response = (FindCoordinatorResponse) abstractResponse;
-        if (batch) {
+        // Handle response based on whether batched version is used by checking `coordinators()` in the response data
+        if (!response.data().coordinators().isEmpty()) {


Both of consumer coordinator and transaction coordinator parse response according to version. By contrast, it check coordinators. Could we unify the behavior? Could consumer coordinator and transaction coordinator check coordinators instead? It seems checking coordinators can simplify the code as we don't need to pass request version to each handler.

chia7712 · 2021-07-04T13:51:18Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

-        } else {
-            data.setKey(coordinatorKey);
-        }
+        data.setKey(coordinatorKey);


ditto (code style)

dajac

@rajinisivaram Thanks for the fix! I left few suggestions.

dajac · 2021-07-05T08:24:07Z

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

        public void onSuccess(ClientResponse resp, RequestFuture<Void> future) {
            log.debug("Received FindCoordinator response {}", resp);

+            boolean batch = resp.requestHeader().apiVersion() >= FindCoordinatorRequest.MIN_BATCHED_VERSION;


How about adding a coordinators method to FindCoordinatorResponse which would either return the list of coordinators (data.coordinators()) if not empty or would return a list containing a Coordinator created from the top level information. That would remove all the batch checks below.

dajac · 2021-07-05T08:29:11Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

+        public void handleResponse(AbstractResponse response, short requestVersion) {
            FindCoordinatorResponse findCoordinatorResponse = (FindCoordinatorResponse) response;
            CoordinatorType coordinatorType = CoordinatorType.forId(builder.data().keyType());
+            boolean batchFindCoordinator = requestVersion >= FindCoordinatorRequest.MIN_BATCHED_VERSION;


The above suggestion would also us to avoid having to pass the requestVersion down here.

dajac · 2021-07-05T08:29:49Z

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

+            env.kafkaClient().setNodeApiVersions(
+                    NodeApiVersions.create(Arrays.asList(findCoordinatorV3, describeGroups)));
+
+            // dummy response for MockClient to handle the UnsupportedVersionException correctly to switch from batched to un-batched


nit: dummy -> Dummy?

dajac · 2021-07-05T08:30:07Z

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

+
+            // dummy response for MockClient to handle the UnsupportedVersionException correctly to switch from batched to un-batched
+            env.kafkaClient().prepareResponse(null);
+            //Retriable FindCoordinatorResponse errors should be retried


nit: Should we add a space before Retriable?

rajinisivaram · 2021-07-05T10:32:14Z

@chia7712 @dajac Thanks for the reviews, addressed comments.

dajac

LGTM, assuming tests pass. Thanks!

chia7712

LGTM

rajinisivaram · 2021-07-05T13:24:01Z

@showuon @chia7712 @dajac Thanks for the reviews. Test failure (RaftClusterTest.testCreateClusterAndCreateAndManyTopics) not related. Merging to trunk.

…on version (apache#10965) KIP-699 added support for batching in FindCoordinatorRequest using a new protocol that changes the wire format for both batched and unbatched requests. Clients were updated to try the new format first and switch irreversibly to the old format if the new format is not supported on one broker. During rolling upgrade (or a downgrade), it is possible that a broker doesn't support new format at some point while other brokers do at a later point. Clients end up in a bad state until restarted since they use new version with old format. This PR changes FindCoordinatorRequest to set data based on actual version when a single group is used. This is always the case for consumer coordinator and transaction manager. For admin API, we still switch to unbatched mode on failure, but the data is set based on actual version, so we never fail even if brokers are upgraded/downgraded. Reviewers: Luke Chen <showuon@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>, David Jacot <djacot@confluent.io>

rajinisivaram requested review from dajac and mimaison July 3, 2021 10:33

KAFKA-13029; Set appropriate fields for FindCoordinatorRequest based …

59adf86

…on version

rajinisivaram force-pushed the KAFKA-13029-coordinator-batch branch from ba2bee9 to 59adf86 Compare July 3, 2021 12:16

showuon reviewed Jul 4, 2021

View reviewed changes

Address review comments

6ab08c2

mimaison mentioned this pull request Jul 4, 2021

KAFKA-13030 FindCoordinators batching causes slow poll when requestin… #10963

Closed

3 tasks

mimaison approved these changes Jul 4, 2021

View reviewed changes

chia7712 reviewed Jul 4, 2021

View reviewed changes

dajac reviewed Jul 5, 2021

View reviewed changes

Address review comments

ee4e4ba

dajac approved these changes Jul 5, 2021

View reviewed changes

chia7712 approved these changes Jul 5, 2021

View reviewed changes

rajinisivaram merged commit 2b6365c into apache:trunk Jul 5, 2021


		public class FindCoordinatorRequest extends AbstractRequest {

		public static final short MIN_BATCHED_VERSION = 4;

KAFKA-13029; Set appropriate fields for FindCoordinatorRequest based on version #10965

KAFKA-13029; Set appropriate fields for FindCoordinatorRequest based on version #10965

Uh oh!

Conversation

rajinisivaram commented Jul 3, 2021

Committer Checklist (excluded from commit message)

Uh oh!

showuon left a comment

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!

rajinisivaram commented Jul 4, 2021

Uh oh!

mimaison left a comment

Choose a reason for hiding this comment

Uh oh!

chia7712 left a comment

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!

dajac left a comment

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!

rajinisivaram commented Jul 5, 2021

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

rajinisivaram commented Jul 5, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants