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

dajac · 2022-11-23T16:26:20Z

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

Committer Checklist (excluded from commit message)

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

jolshan · 2022-12-21T22:30:48Z

reminder to rebase here :)

jolshan · 2022-12-27T22:33:25Z

also minor nit in the title TnxOffsetCommit -> TxnOffsetCommit

…rface

dajac · 2023-01-09T16:08:40Z

@jolshan I have updated the PR.

jolshan · 2023-01-10T21:26:14Z

+        sendResponse(responseBuilder.build())
+        CompletableFuture.completedFuture[Unit](())
+      } else {
+        val txnOffsetCommitRequestData = new TxnOffsetCommitRequestData()


Are we only replacing the topics here? There's a lot of code to move the majority of one request data into a new one.

Right. It is basically a copy of the original request but with the authorized/existing topics.

i agree, it doesn't seem ideal. what's the reason for not reusing the original request data and applying setTopics()?

We cannot mutate the original request. In case of an error resulting in calling txnOffsetCommitRequest.getErrorResponse(exception), the error response would only include the authorised topics and not all of the original one.

I agree that it is not ideal. I considered a few alternatives:

We could use .duplicate().setTopics(...) but duplicate also copy the original topics and that it is not necessary;

We could change the GroupCoordinator interface to take individual arguments but I thought that keeping the interface consistent is better.

In the end, I went with the manual copy.

jolshan · 2023-01-10T21:42:03Z

+        partitions += tp -> new OffsetAndMetadata(
+          offset = partition.committedOffset,
+          leaderEpoch = partition.committedLeaderEpoch match {
+            case RecordBatch.NO_PARTITION_LEADER_EPOCH => Optional.empty[Integer]


Did we previously not handle -1 correctly here? In KafkaApis 2544 we just keep -1.

It was handled here in the previous code.

Oh didn't see this. Thanks for sharing 😅 I like that both are handled in the same place now.

jolshan · 2023-01-10T21:47:14Z

  def testJoinGroup(version: Short): Unit = {
    val groupCoordinator = mock(classOf[GroupCoordinator])
-    val adapter = new GroupCoordinatorAdapter(groupCoordinator)
+    val adapter = new GroupCoordinatorAdapter(groupCoordinator, Time.SYSTEM)


Curious if we wanted to mock time here. Doesn't seem to be a big difference one way or the for most of these since they don't use a time component anyway, but could be a good practice to have a MockTime. (Looks like we do use it for the test added 😄 )

Why? I usually use MockTime only when I need it. Otherwise, I default to Time.SYSTEM.

Maybe I'm just recalling some recent tests where I used system time because it was already there and it led to flakiness.
I suppose here though it is easier to switch so maybe not a problem.

Yeah. In this case, it does not matter because the test does not depend on time at all.

jolshan · 2023-01-10T22:19:59Z

+          requestLocal.bufferSupplier
+        ).handle[Unit] { (response, exception) =>
+          if (exception != null) {
+            sendResponse(txnOffsetCommitRequest.getErrorResponse(exception))


Did we want to still send the existing responsebuilder response at all or mark the whole request as failed?

Ah -- I misunderstood this error. This is how we handle errors that would have previously been thrown back to KafkaApis.

I guess my confusion is when we see this line vs. in the handleError above.

Yeah, this idea is to catch unexpected errors here. Otherwise, the future would return a response. Note that handleError is here to handle errors thrown in the handle. It is basically a safety net.

Ah -- so we typically don't expect handleError to see anything since this handle should get most of the errors.

That's right. handleError is more for unexpected errors.

jolshan · 2023-01-11T00:01:47Z

+  }
+
+  @Test
+  def testHandleTxnOffsetCommitRequestTopicsAndPartitionsValidation(): Unit = {


This test is super thorough! Thanks for adding

jolshan

I've left a few comments. Looks pretty good so far.

…mmit-offset

dajac · 2023-01-12T17:49:21Z

@jolshan @jeffkbkim Updated the PR.

jolshan · 2023-01-12T18:20:48Z

-          topic = new TxnOffsetCommitResponseData.TxnOffsetCommitResponseTopic().setName(tp.topic)
-          byTopics.put(tp.topic, topic)
-          response.topics.add(topic)
+        val topic = byTopics.get(tp.topic) match {


This seems like it could use the getOrElseUpdate method, but I'm not sure if we typically have side effects like adding to the response. Not a huge deal either way.

Yeah, I didn’t use it because of the side effect.

The downside is double-lookup (one for get and another for insert).

jolshan

Thanks David. Failures look unrelated.

jeffkbkim

LGTM, left a small comment.

jeffkbkim · 2023-01-13T04:59:48Z

+        sendResponse(responseBuilder.build())
+        CompletableFuture.completedFuture[Unit](())
+      } else {
+        val txnOffsetCommitRequestData = new TxnOffsetCommitRequestData()


i agree, it doesn't seem ideal. what's the reason for not reusing the original request data and applying setTopics()?

dajac · 2023-01-13T07:51:27Z

Failed tests are not related.

…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) ...

…rface (apache#12901) This patch adds `TxnOffsetCommit` to the new `GroupCoordinator` interface and updates `KafkaApis` to use it. Reviewers: 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 23, 2022

dajac changed the title ~~KAFKA-14367; Add TnxOffsetCommit to the new GroupCoordinator interface~~ KAFKA-14367; Add TxnOffsetCommit to the new GroupCoordinator interface Jan 9, 2023

dajac added 2 commits January 9, 2023 16:54

KAFKA-14367; Add TnxOffsetCommit to the new GroupCoordinator inte…

e7060bd

…rface

refactor

a8d5423

dajac force-pushed the KAFKA-14367-txn-commit-offset branch from 10503c9 to a8d5423 Compare January 9, 2023 16:07

dajac requested a review from jolshan January 9, 2023 16:08

jolshan reviewed Jan 10, 2023

View reviewed changes

jolshan reviewed Jan 11, 2023

View reviewed changes

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

dfad072

…mmit-offset

jolshan reviewed Jan 11, 2023

View reviewed changes

Comment thread core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorAdapterTest.scala Outdated

dajac added 3 commits January 12, 2023 16:24

fixup

6d17c95

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

301b447

…mmit-offset

refactor

0d14658

jolshan reviewed Jan 12, 2023

View reviewed changes

jolshan approved these changes Jan 12, 2023

View reviewed changes

jeffkbkim approved these changes Jan 13, 2023

View reviewed changes

dajac merged commit a2926ed into apache:trunk Jan 13, 2023

dajac deleted the KAFKA-14367-txn-commit-offset branch January 13, 2023 08:55

Conversation

dajac commented Nov 23, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

jolshan commented Dec 21, 2022

Uh oh!

jolshan commented Dec 27, 2022

Uh oh!

dajac commented Jan 9, 2023

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!

jolshan Jan 10, 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!

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!

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!

jolshan left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

dajac commented Jan 12, 2023

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!

jolshan left a comment

Choose a reason for hiding this comment

Uh oh!

jeffkbkim left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dajac commented Jan 13, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

dajac commented Nov 23, 2022 •

edited

Loading

jolshan Jan 10, 2023 •

edited

Loading