KAFKA-14367; Add `OffsetFetch` to the new `GroupCoordinator` interface by dajac · Pull Request #12870 · apache/kafka

dajac · 2022-11-17T14:32:51Z

This patch adds OffsetFetch to the new GroupCoordinator interface and updates KafkaApis to use it. The changes in KafkaApis are larger than what I was hoping for. I think that we should refactor this part of the code even further but I leave this for a further PR.

Committer Checklist (excluded from commit message)

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

dajac · 2022-11-17T17:15:22Z

I have to add a few unit tests in KafkaApisTest. There are none at the moment.

dajac · 2022-11-21T09:05:48Z

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

I have removed this test because I think that this is wrong. When the same group id is specified multiple times, we should rather throw an invalid request rather than doing this. Thoughts?

This was basically due to the fact that we were using an hash map so the last occurrence of a group won. We don't use the hashmap anymore with the patch. Instead, we return an entry for every group present in the request.

Was this behavior documented or just a quirk of the implementation?

I'm also curious if making this change (invalid requests) will break someone's clients.

I think that it is a quirk of the implementation. It is because we used HashMaps before this patch so the last one wins.

In my opinion, it is pretty bad to just ignore one group passed in the request like this.

Are we deciding to actually throw the error? If so we should document that.

Otherwise, maybe we can add a comment in the request file that requesting the group multiple times in the request will also give us the response multiple times as of this release.

I need to look into other APIs to see how we usually handle this. At a first glance, it seems that we are not consistent.

Did we ever get a conclusion here? If so let's just document what we decided.

+1 for having duplicate responses if a group is repeated. i don't think this warrants an invalid request but we should conform to the norm if there is one.

I have looked at other APIs and we are not consistent, unfortunately. I believe that my current implementation (returning a response for each provided group in the same order) is the right way and likely the expected way. We could consider this as a bug in the KIP-709 implementation. I have asked about this in the KIP-709 discuss thread as well.

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

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java

jolshan · 2022-12-20T22:16:51Z

Was looking for the tests and then remembered this comment:
#12870 (comment)

Reminder on this to add the tests 😄

…offset

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

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

dajac · 2022-12-21T15:06:35Z

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

@@ -19,6 +19,7 @@
 import java.util.Map.Entry;


As I follow-up, I would like to clean this class. There are way too many ways to construct this object and the logic is pretty complicated. It is the same for the request. Let's do this separately.

sounds like a good plan. Thanks!

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

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

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

jeffkbkim

thanks for the PR! the code looks much more simplified and cleaner. left some comments.

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

jeffkbkim · 2023-01-06T23:39:58Z

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

+                        newPartition = new OffsetFetchResponsePartition()
+                            .setPartitionIndex(partition.partitionIndex())
+                            .setErrorCode(group.errorCode());
+                    } else {


i'm a bit confused on the else statement. we can hit this either when version >= 2 or group error == NONE.

based on the comment above, it seems that for version >= 2 we don't have to put error at the partition level. also, do we need to add offset/metadata if there is an error for version >= 2?

It is still possible to have a partition level error with version >= 2 (e.g. UNSTABLE_OFFSET_COMMIT). To answer your second point, if there is an error, the offset/metadata should be correctly set at this stage so we can just copy whatever we have got here.

that makes sense. thanks for the clarification

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

jeffkbkim · 2023-01-07T00:16:46Z

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

-            DESCRIBE, TOPIC, allPartitionData)(_.topic)
-          (Errors.NONE, authorizedPartitionData)
-        }
+  private def fetchAllOffsets(


fetchAllOffsetsForGroup and fetchOffsetsForGroup makes it more readable for me. wdyt?

looking at the new GroupCoordinator interface seems like they have a counterpart method. thoughts on changing both?

Renaming in KafkaApis is reasonable. For GroupCoordinator, the ForGroup seems a bit redundant so I would rather keep it as it is there.

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

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

jeffkbkim · 2023-01-07T01:03:57Z

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

+1 for having duplicate responses if a group is repeated. i don't think this warrants an invalid request but we should conform to the norm if there is one.

dajac · 2023-01-09T14:53:03Z

@jolshan @jeffkbkim I have updated the PR. Could you take another look?

…offset

jeffkbkim

LGTM if tests pass

jolshan · 2023-01-09T23:22:58Z

Looks fairly reasonable. I'm going to rebuild though to see if some of the tests look flaky.

dajac · 2023-01-10T09:51:23Z

Failed tests seem unrelated:

Build / JDK 8 and Scala 2.12 / testValidSaslPlainOverPlaintext() – org.apache.kafka.common.security.authenticator.SaslAuthenticatorTest
<1s
Build / JDK 8 and Scala 2.12 / shouldPauseStandbyTaskAndNotTransitToUpdateStandbyAgain() – org.apache.kafka.streams.processor.internals.DefaultStateUpdaterTest
15s
Build / JDK 17 and Scala 2.13 / testOneWayReplicationWithAutoOffsetSync() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsWithCustomForwardingAdminIntegrationTest
3m 23s
Build / JDK 17 and Scala 2.13 / testTaskRequestWithOldStartMsGetsUpdated() – org.apache.kafka.trogdor.coordinator.CoordinatorTest
2m 0s
Build / JDK 11 and Scala 2.13 / testLargeAssignmentAndGroupWithNonEqualSubscription() – org.apache.kafka.clients.consumer.StickyAssignorTest
1m 3s
Build / JDK 11 and Scala 2.13 / testReplication() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationSSLTest

jolshan

Thanks for the hard work David :)

* apache-github/trunk: KAFKA-14601: Improve exception handling in KafkaEventQueue apache#13089 KAFKA-14367; Add `OffsetCommit` to the new `GroupCoordinator` interface (apache#12886) KAFKA-14530: Check state updater more often (apache#13017) KAFKA-14304 Use boolean for ZK migrating brokers in RPC/record (apache#13103) KAFKA-14003 Kafka Streams JUnit4 to JUnit5 migration part 2 (apache#12301) KAFKA-14607: Move Scheduler/KafkaScheduler to server-common (apache#13092) KAFKA-14367; Add `OffsetFetch` to the new `GroupCoordinator` interface (apache#12870) KAFKA-14557; Lock metadata log dir (apache#13058) MINOR: Implement toString method for TopicAssignment and PartitionAssignment (apache#13101) KAFKA-12558: Do not prematurely mutate internal partition state in Mirror Maker 2 (apache#11818) KAFKA-14540: Fix DataOutputStreamWritable#writeByteBuffer (apache#13032) KAFKA-14600: Reduce flakiness in ProducerIdExpirationTest (apache#13087) KAFKA-14279: Add 3.3.x streams system tests (apache#13077) MINOR: bump streams quickstart pom versions and add to list in gradle.properties (apache#13064) MINOR: Update KRaft cluster upgrade documentation for 3.4 (apache#13063) KAFKA-14493: Introduce Zk to KRaft migration state machine STUBs in KRaft controller. (apache#12998) KAFKA-14570: Fix parenthesis in verifyFullFetchResponsePartitions output (apache#13072) MINOR: Remove public mutable fields from ProducerAppendInfo (apache#13091)

…master * apache-github/trunk: (23 commits) MINOR: Include the inner exception stack trace when re-throwing an exception (apache#12229) MINOR: Fix docs to state that sendfile implemented in `TransferableRecords` instead of `MessageSet` (apache#13109) Update ProducerConfig.java (apache#13115) KAFKA-14618; Fix off by one error in snapshot id (apache#13108) KAFKA-13709 (follow-up): Avoid mention of 'exactly-once delivery' or 'delivery guarantees' in Connect (apache#13106) KAFKA-14367; Add `TxnOffsetCommit` to the new `GroupCoordinator` interface (apache#12901) KAFKA-14568: Move FetchDataInfo and related to storage module (apache#13085) KAFKA-14612: Make sure to write a new topics ConfigRecords to metadata log iff the topic is created (apache#13104) KAFKA-14601: Improve exception handling in KafkaEventQueue apache#13089 KAFKA-14367; Add `OffsetCommit` to the new `GroupCoordinator` interface (apache#12886) KAFKA-14530: Check state updater more often (apache#13017) KAFKA-14304 Use boolean for ZK migrating brokers in RPC/record (apache#13103) KAFKA-14003 Kafka Streams JUnit4 to JUnit5 migration part 2 (apache#12301) KAFKA-14607: Move Scheduler/KafkaScheduler to server-common (apache#13092) KAFKA-14367; Add `OffsetFetch` to the new `GroupCoordinator` interface (apache#12870) KAFKA-14557; Lock metadata log dir (apache#13058) MINOR: Implement toString method for TopicAssignment and PartitionAssignment (apache#13101) KAFKA-12558: Do not prematurely mutate internal partition state in Mirror Maker 2 (apache#11818) KAFKA-14540: Fix DataOutputStreamWritable#writeByteBuffer (apache#13032) KAFKA-14600: Reduce flakiness in ProducerIdExpirationTest (apache#13087) ...

apache#12870) This patch adds OffsetFetch to the new GroupCoordinator interface and updates KafkaApis to use it. Reviewers: Philip Nee <pnee@confluent.i>, Jeff Kim <jeff.kim@confluent.io>, Justine Olshan <jolshan@confluent.io>

dajac added the KIP-848 The Next Generation of the Consumer Rebalance Protocol label Nov 17, 2022

dajac force-pushed the KAFKA-14367-fetch-offset branch from bfe7da5 to df9ee42 Compare November 17, 2022 14:43

dajac commented Nov 21, 2022

View reviewed changes

KAFKA-14367; Add OffsetFetch to the new GroupCoordinator interface

4b1b652

dajac force-pushed the KAFKA-14367-fetch-offset branch from 222da0e to 4b1b652 Compare December 16, 2022 14:27

dajac changed the base branch from KAFKA-14367-join-group to trunk December 16, 2022 14:27

refactor

d15dbdc

jolshan reviewed Dec 20, 2022

View reviewed changes

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

jolshan reviewed Dec 20, 2022

View reviewed changes

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

jolshan reviewed Dec 20, 2022

View reviewed changes

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java Outdated Show resolved Hide resolved

dajac added 6 commits December 21, 2022 12:10

refactor to simplify code in KafkaApis

15a1a16

add unit tests

307b585

small cleanup

4258cbf

add test

3f80d6f

Merge remote-tracking branch 'upstream/trunk' into KAFKA-14367-fetch-…

ecd893e

…offset

fixes to accomote trunk changes

23fbbb1

dajac commented Dec 21, 2022

View reviewed changes

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

dajac commented Dec 21, 2022

View reviewed changes

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

dajac commented Dec 21, 2022

View reviewed changes

dajac requested a review from jolshan December 21, 2022 15:10

philipnee reviewed Dec 21, 2022

View reviewed changes

jeffkbkim reviewed Jan 7, 2023

View reviewed changes

address minor comments

31f843f

dajac requested review from jeffkbkim and philipnee and removed request for jeffkbkim and philipnee January 9, 2023 14:52

Merge remote-tracking branch 'upstream/trunk' into KAFKA-14367-fetch-…

6f913d7

…offset

jeffkbkim approved these changes Jan 9, 2023

View reviewed changes

jolshan approved these changes Jan 10, 2023

View reviewed changes

jolshan merged commit 24a8642 into apache:trunk Jan 10, 2023

dajac deleted the KAFKA-14367-fetch-offset branch January 11, 2023 10:23

Conversation

dajac commented Nov 17, 2022

Committer Checklist (excluded from commit message)

Uh oh!

dajac commented Nov 17, 2022

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dajac Jan 9, 2023 • 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!

Uh oh!

Uh oh!

Uh oh!

jolshan commented Dec 20, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

jeffkbkim 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dajac commented Jan 9, 2023

Uh oh!

jeffkbkim left a comment

Choose a reason for hiding this comment

Uh oh!

jolshan commented Jan 9, 2023

Uh oh!

dajac commented Jan 10, 2023

Uh oh!

jolshan left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

dajac Jan 9, 2023 •

edited

Loading

jolshan commented Dec 20, 2022 •

edited

Loading