KAFKA-14367; Add `JoinGroup` to the new `GroupCoordinator` interface #12845

dajac · 2022-11-14T08:35:16Z

This patch adds joinGroup to the new GroupCoordinator interface and updates KafkaApis to use it.

For the context, I will do the same for all the other interactions with the current group coordinator. In order to limit the changes, I have chosen to introduce the GroupCoordinatorAdapter that translates the new interface to the old one. It is basically a wrapper. This allows keeping the current group coordinator untouched for now and focus on the KafkaApis changes. Eventually, we can remove GroupCoordinatorAdapter.

Committer Checklist (excluded from commit message)

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

core/src/main/scala/kafka/server/KafkaApis.scala

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

core/src/main/scala/kafka/server/BrokerServer.scala

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

core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorAdapterTest.scala

jolshan · 2022-11-15T21:28:14Z

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

-      testJoinGroupProtocolType(version.asInstanceOf[Short])
-    }
-  }
+  def testHandleJoinGroupRequestFutureFailed(): Unit = {


Is this also a new test?

That's right.

jolshan

seems reasonable so far. This is just an initial pass. I will take another pass soon.

jeffkbkim

thanks for the PR, left some comments.

is the idea to add all existing group coordinator apis to GroupCoordinatorAdapter, then introduce the new GroupCoordinator which implements GroupCoordinator interface with new design, then remove GroupCoordinatorAdapter and point KafkaApis to the new GroupCoordinator?

core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala

core/src/main/scala/kafka/server/KafkaApis.scala

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorAdapterTest.scala

core/src/main/scala/kafka/server/KafkaApis.scala

dajac · 2022-11-16T09:39:31Z

@jeffkbkim @jolshan Thanks for your comments. I have addressed your feedback. Note that I have extracted the changes about the JoinGroupResponse version handling in a separate PR: #12864. It is less risky this way.

…ty in JoinGroupResponse (#12864) This is a small refactor extracted from #12845. It basically moves the logic to handle the backward compatibility of `JoinGroupResponseData.protocolName` from `KafkaApis` to `JoinGroupResponse`. The patch adds a new unit test for `JoinGroupResponse` and relies on existing tests as well. Reviewers: Justine Olshan <jolshan@confluent.io>, Jason Gustafson <jason@confluent.io>

Add GroupCoordinator interface

dajac · 2022-11-17T08:20:05Z

Rebased the PR. Ready for second round.

jeffkbkim

thanks! left some comments.

core/src/main/scala/kafka/server/KafkaApis.scala

jeffkbkim · 2022-11-18T05:51:04Z

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

+      ArgumentMatchers.eq(joinGroupRequest)
+    )).thenReturn(future)
+
+    val response = new AtomicReference[JoinGroupResponse]()


can you help me understand the reason for using an atomic reference?

i just needed a container to store the response. i could have used a var initialized to null as well, i suppose.

jeffkbkim · 2022-11-18T05:55:53Z

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

+    createKafkaApis().handle(
+      requestChannelRequest,
+      RequestLocal.NoCaching
+    )


i'm surely missing something here - shouldn't the test be blocked here? i can't seem to find the other thread running

it does not. keep in mind that we use a future internally so the method is executed and returns immediately. then, the future is completed later on by future.completeExceptionally(Errors.NOT_COORDINATOR.exception).

gotcha. i think i was missing this from the CompletableFuture java docs

Actions supplied for dependent completions of non-async methods may be performed by the thread that completes the current CompletableFuture, or by any other caller of a completion method.

so the thread that completes the future will attempt to finish all (non-async) dependent chains

dajac · 2022-11-18T07:35:32Z

@jeffkbkim Thanks for the review. I have addressed your comments.

core/src/main/scala/kafka/server/KafkaServer.scala

core/src/main/scala/kafka/server/KafkaApis.scala

jeffkbkim · 2022-11-19T04:42:33Z

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

+    createKafkaApis().handle(
+      requestChannelRequest,
+      RequestLocal.NoCaching
+    )


gotcha. i think i was missing this from the CompletableFuture java docs

Actions supplied for dependent completions of non-async methods may be performed by the thread that completes the current CompletableFuture, or by any other caller of a completion method.

so the thread that completes the future will attempt to finish all (non-async) dependent chains

core/src/main/scala/kafka/server/KafkaApis.scala

jolshan · 2022-11-21T19:56:40Z

@dajac This is looking close to ready. I just had a few points.

I left a comment about leaving more comments on tests, not a big deal either way, but just wanted to remind.
I see a lot of mirror maker tests failing that I couldn't find on other branches. It is likely unrelated, but wanted to confirm.

jeffkbkim

LGTM, left a minor comment. the mirror maker test failures only exist in the last commit, so i think we're good.

jeffkbkim · 2022-11-22T03:56:57Z

core/src/main/scala/kafka/server/KafkaApis.scala

@@ -184,6 +195,7 @@ class KafkaApis(val requestChannel: RequestChannel,
        case ApiKeys.OFFSET_FETCH => handleOffsetFetchRequest(request)
        case ApiKeys.FIND_COORDINATOR => handleFindCoordinatorRequest(request)
        case ApiKeys.JOIN_GROUP => handleJoinGroupRequest(request, requestLocal)
+          .exceptionally(handleError)


nit: i personally think it's neater to merge this with the line above, since all the other cases are single-lined

showuon

Overall LGTM! I understand why you introduce GroupCoordinatorAdapter, to focus on the KafkaApi changes, but I don't think that would have less change. However, I don't have strong preference for it. Let's keep on it! :)

And thanks @jeffkbkim @jolshan for the review.

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

...dinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRequestContext.java

dajac · 2022-11-22T14:41:40Z

I left a comment about leaving more comments on tests, not a big deal either way, but just wanted to remind.

@jolshan Thanks. I thought about this and I feel like the test is self explanatory.

I understand why you introduce GroupCoordinatorAdapter, to focus on the KafkaApi changes, but I don't think that would have less change

@showuon Yeah... The point that convinced me to do it like this was the impact on the tests of the group coordinator. The diff was quite large there.

dajac · 2022-11-22T14:42:17Z

@jeffkbkim @jolshan @showuon Thanks for your comments. I have addressed them.

jolshan

lgtm -- thanks David!

...dinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRequestContext.java

dajac · 2022-11-23T09:06:17Z

I have addressed Jason's comments.

hachikuji

Thanks, LGTM

* apache-kafka/trunk: (31 commits) KAFKA-14413: Separate MirrorMaker configurations for each connector (apache#12899) MINOR: Fix config documentation formatting (apache#12921) KAFKA-13731: Allow standalone workers to be started without providing any connector configurations (apache#11890) MINOR: Fix docs in security.html (apache#12851) MINOR: Remove config/kraft/README.md from rat exclusion list (apache#12923) MINOR: Update Gradle to 7.6 (apache#12918) KAFKA-14414: Fix request/response header size calculation (apache#12917) KAFKA-14339 : Do not perform producerCommit on serializationError when trying offsetWriter flush (apache#12920) KAFKA-14367; Add `JoinGroup` to the new `GroupCoordinator` interface (apache#12845) MINOR: buildoutput.log should be under the `build` directory (apache#12919) ...

…ty in JoinGroupResponse (apache#12864) This is a small refactor extracted from apache#12845. It basically moves the logic to handle the backward compatibility of `JoinGroupResponseData.protocolName` from `KafkaApis` to `JoinGroupResponse`. The patch adds a new unit test for `JoinGroupResponse` and relies on existing tests as well. Reviewers: Justine Olshan <jolshan@confluent.io>, Jason Gustafson <jason@confluent.io>

…pache#12845) This patch adds `joinGroup` to the new `GroupCoordinator` interface and updates `KafkaApis` to use it. For the context, I will do the same for all the other interactions with the current group coordinator. In order to limit the changes, I have chosen to introduce the `GroupCoordinatorAdapter` that translates the new interface to the old one. It is basically a wrapper. This allows keeping the current group coordinator untouched for now and focus on the `KafkaApis` changes. Eventually, we can remove `GroupCoordinatorAdapter`. Reviewers: Justine Olshan <jolshan@confluent.io>, Jeff Kim <jeff.kim@confluent.io>, Luke Chen <showuon@gmail.com>, Jason Gustafson <jason@confluent.io>

dajac commented Nov 14, 2022

View reviewed changes

core/src/main/scala/kafka/server/KafkaApis.scala Show resolved Hide resolved

dajac commented Nov 14, 2022

View reviewed changes

core/src/main/scala/kafka/server/KafkaApis.scala Show resolved Hide resolved

dajac commented Nov 14, 2022

View reviewed changes

core/src/main/scala/kafka/server/KafkaApis.scala Show resolved Hide resolved

dajac commented Nov 14, 2022

View reviewed changes

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala Show resolved Hide resolved

dajac commented Nov 14, 2022

View reviewed changes

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala Show resolved Hide resolved

dajac commented Nov 14, 2022

View reviewed changes

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala Show resolved Hide resolved

dajac commented Nov 14, 2022

View reviewed changes

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala Show resolved Hide resolved

dajac force-pushed the KAFKA-14367-join-group branch from e4e8ceb to dbe20a5 Compare November 14, 2022 08:49

dajac added the KIP-848 label Nov 14, 2022

jolshan reviewed Nov 15, 2022

View reviewed changes

core/src/main/scala/kafka/server/BrokerServer.scala Show resolved Hide resolved

jolshan reviewed Nov 15, 2022

View reviewed changes

clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java Outdated Show resolved Hide resolved

jolshan reviewed Nov 15, 2022

View reviewed changes

core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorAdapterTest.scala Show resolved Hide resolved

jolshan reviewed Nov 15, 2022

View reviewed changes

jeffkbkim reviewed Nov 16, 2022

View reviewed changes

dajac mentioned this pull request Nov 16, 2022

MINOR: Handle JoinGroupResponseData.protocolName backward compatibility in JoinGroupResponse #12864

Merged

3 tasks

dajac force-pushed the KAFKA-14367-join-group branch from dbe20a5 to 83b72f6 Compare November 16, 2022 09:37

dajac added 2 commits November 17, 2022 08:34

KAFKA-14367; Add JoinGroup to the new GroupCoordinator interface

57736b4

Add GroupCoordinator interface

address comments

01baae5

dajac force-pushed the KAFKA-14367-join-group branch from 83b72f6 to 01baae5 Compare November 17, 2022 07:35

Handle unexpected exception

33fa71d

jeffkbkim reviewed Nov 18, 2022

View reviewed changes

address feedback

4aed39e

jolshan reviewed Nov 18, 2022

View reviewed changes

core/src/main/scala/kafka/server/KafkaServer.scala Outdated Show resolved Hide resolved

jolshan reviewed Nov 18, 2022

View reviewed changes

core/src/main/scala/kafka/server/KafkaApis.scala Show resolved Hide resolved

jeffkbkim reviewed Nov 19, 2022

View reviewed changes

fix imports

75a3a44

jeffkbkim approved these changes Nov 22, 2022

View reviewed changes

showuon approved these changes Nov 22, 2022

View reviewed changes

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala Show resolved Hide resolved

...dinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRequestContext.java Outdated Show resolved Hide resolved

address comments

a63e91d

jolshan approved these changes Nov 22, 2022

View reviewed changes

hachikuji reviewed Nov 22, 2022

View reviewed changes

...dinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRequestContext.java Outdated Show resolved Hide resolved

...dinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRequestContext.java Outdated Show resolved Hide resolved

address Jason's comments

4649e29

fix checkstyle

418c880

hachikuji approved these changes Nov 29, 2022

View reviewed changes

dajac merged commit 98e19b3 into apache:trunk Nov 29, 2022

dajac deleted the KAFKA-14367-join-group branch November 29, 2022 20:06

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-14367; Add `JoinGroup` to the new `GroupCoordinator` interface #12845

KAFKA-14367; Add `JoinGroup` to the new `GroupCoordinator` interface #12845

dajac commented Nov 14, 2022

jolshan Nov 15, 2022

dajac Nov 16, 2022

jolshan left a comment

jeffkbkim left a comment

dajac commented Nov 16, 2022

dajac commented Nov 17, 2022

jeffkbkim left a comment

jeffkbkim Nov 18, 2022

dajac Nov 18, 2022

jeffkbkim Nov 18, 2022

dajac Nov 18, 2022

jeffkbkim Nov 19, 2022

dajac commented Nov 18, 2022

jeffkbkim Nov 19, 2022

jolshan commented Nov 21, 2022

jeffkbkim left a comment

jeffkbkim Nov 22, 2022

showuon left a comment

dajac commented Nov 22, 2022 •

edited

Loading

dajac commented Nov 22, 2022

jolshan left a comment

dajac commented Nov 23, 2022

hachikuji left a comment

KAFKA-14367; Add JoinGroup to the new GroupCoordinator interface #12845

KAFKA-14367; Add JoinGroup to the new GroupCoordinator interface #12845

Conversation

dajac commented Nov 14, 2022

Committer Checklist (excluded from commit message)

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jolshan left a comment

Choose a reason for hiding this comment

jeffkbkim left a comment

Choose a reason for hiding this comment

dajac commented Nov 16, 2022

dajac commented Nov 17, 2022

jeffkbkim 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

dajac commented Nov 18, 2022

Choose a reason for hiding this comment

jolshan commented Nov 21, 2022

jeffkbkim left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

showuon left a comment

Choose a reason for hiding this comment

dajac commented Nov 22, 2022 • edited Loading

dajac commented Nov 22, 2022

jolshan left a comment

Choose a reason for hiding this comment

dajac commented Nov 23, 2022

hachikuji left a comment

Choose a reason for hiding this comment

KAFKA-14367; Add `JoinGroup` to the new `GroupCoordinator` interface #12845

KAFKA-14367; Add `JoinGroup` to the new `GroupCoordinator` interface #12845

dajac commented Nov 22, 2022 •

edited

Loading