KAFKA-18060: new coordinator does not handle TxnOffsetCommitRequest with empty member id when using CONSUMER group #17914

brandboat · 2024-11-22T15:00:17Z

There are two issues in KAFKA-18060:

New coordinator can't handle the TxnOffsetCommitRequest with empty member id, and TxnOffsetCommitRequest v0-v2 do definitely has an empty member ID, causing ConsumerGroup#validateOffsetCommit to throw an UnknownMemberIdException. This prevents the old producer from calling sendOffsetsToTransaction. Note that TxnOffsetCommitRequest versions v0-v2 are included in KIP-896, so it seems the new coordinator should support that operations
The deprecated API Producer#sendOffsetsToTransaction does not use v0-v2 to send TxnOffsetCommitRequest with an empty member ID. Unfortunately, it has been released for a while. Therefore, the new coordinator needs to handle TxnOffsetCommitRequest with an empty member ID for all versions.

Taken from the two issues above, we need to handle empty member id in all API versions when new coordinator are dealing with TxnOffsetCommitRequest.

Committer Checklist (excluded from commit message)

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

…th empty member id when using CONSUMER group

dajac · 2024-11-22T15:54:03Z

@brandboat Thanks for the quick patch. Could you please add a description to the pull request? We usually also use it for the commit message.

dajac

@brandboat Thanks for the patch. I left some comments for consideration.

dajac · 2024-11-22T16:04:50Z

...rdinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java

+        // TxnOffsetCommitRequest versions v0-v2 do not include a member ID.
+        // And we can still send UNKNOWN_GENERATION_ID and UNKNOWN_MEMBER_ID through Producer#sendOffsetsToTransaction.
+        // Therefore, we should not throw exception in these cases.


I would simplify it a bit. What about The TxnOffsetCommit API does not require the member id, the generation id and the group instance id fields. Hence the are only validated if any of them is provided.?

dajac · 2024-11-22T16:08:10Z

...rdinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java

@@ -26,6 +26,7 @@
 import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;


Could you please extend unit tests in ConsumerGroupTest too?

dajac · 2024-11-22T16:11:18Z

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

+      adminClient = cluster.admin()
+      adminClient.createTopics(singleton(new NewTopic(topic, 1, 1.toShort)))


You could actually use createTopic method from the parent class.

dajac · 2024-11-22T16:12:02Z

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

+      // add this line to avoid error INVALID_TXN_STATE
+      producer.sendOffsetsToTransaction(
+        Collections.singletonMap(new TopicPartition("topic", 0), new OffsetAndMetadata(5)),
+        new ConsumerGroupMetadata(groupId))


I suppose that you must do this in order to add the __consumer_offsets partition to the transaction.

Personally, I prefer to build the AddOffsetsToTxnRequest explicitly. for example:

protected def addOffsetsToTxn(groupId: String, producerId: Int, producerEpoch: Short, transactionalId: String, version: Short): Unit = { val request = new AddOffsetsToTxnRequest.Builder(new AddOffsetsToTxnRequestData() .setTransactionalId(transactionalId) .setProducerId(producerId).setProducerEpoch(producerEpoch).setGroupId(groupId) ).build(version) val expectedResponse = new AddOffsetsToTxnResponseData() val response = connectAndReceive[AddOffsetsToTxnResponse](request) assertEquals(expectedResponse, response.data) }

Thank you both! I believe all comments have been addressed. Please take a look when you have a chance.

dajac · 2024-11-22T16:12:29Z

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

+        Collections.singletonMap(new TopicPartition("topic", 0), new OffsetAndMetadata(5)),
+        new ConsumerGroupMetadata(groupId))
+
+      // verify that the TXN_OFFSET_COMMIT request is processed correctly when member id is UNKNOWN_MEMBER_ID


nit: Verify....

dajac · 2024-11-22T16:12:58Z

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

+      producer.close()
+    }
+  }
+}


While we are here, do you mind adding a test which verifies that the fields are correctly validated when provided?

dajac · 2024-11-22T16:13:29Z

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

+      new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+      new ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+      new ClusterConfigProperty(key = TransactionLogConfig.TRANSACTIONS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+      new ClusterConfigProperty(key = TransactionLogConfig.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),


nit: You can define all the common configs at the class level in ClusterTestDefaults.

dajac

@brandboat Thanks for the update. I left a few more comments for consideration.

dajac · 2024-11-26T09:15:42Z

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

+  protected def addOffsetsToTxn(groupId: String,
+                                producerId: Int,
+                                producerEpoch: Short,
+                                transactionalId: String,
+                                version: Short): Unit = {


nit: Let's follow the format of the other methods please.

Sorry, that comes from my auto format... I forgot to set the scala code style, thanks

dajac · 2024-11-26T09:16:03Z

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

+                                version: Short): Unit = {
+    val request = new AddOffsetsToTxnRequest.Builder(new AddOffsetsToTxnRequestData()
+      .setTransactionalId(transactionalId)
+      .setProducerId(producerId).setProducerEpoch(producerEpoch).setGroupId(groupId)


nit: Let's put setProducerEpoch and setGroupId on new lines to be consistent.

dajac · 2024-11-26T09:16:13Z

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

+      .setProducerId(producerId).setProducerEpoch(producerEpoch).setGroupId(groupId)
+    ).build(version)
+
+    val expectedResponse = new AddOffsetsToTxnResponseData()


nit: We could inline it.

dajac · 2024-11-26T09:19:25Z

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

+      producer = cluster.producer(Collections.singletonMap(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId))
+      producer.initTransactions()
+      producer.beginTransaction()


Now that we have addOffsetsToTxn, I wonder if we should also add one to init the transaction so we could remove the producer. It is a bit weird to use both because you actually don't know whether the producer id/epoch passes to addOffsetsToTxn are actually correct.

dajac · 2024-11-26T09:20:06Z

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

+          memberId = JoinGroupRequest.UNKNOWN_MEMBER_ID,
+          generationId = JoinGroupRequest.UNKNOWN_GENERATION_ID,


Could you please add a similar test but which sets member id/generation id too?

dajac · 2024-11-26T09:21:12Z

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

+        TestUtils.waitUntilTrue(() => {
+          consumer.poll(Duration.ofSeconds(5)).asScala
+            .filter(record => record.key() != null && record.value() != null)
+            .map(record => new GroupCoordinatorRecordSerde()
+              .deserialize(ByteBuffer.wrap(record.key().get()), ByteBuffer.wrap(record.value().get())))
+            .filter(coordinatorRecord =>
+              coordinatorRecord.key().message().isInstanceOf[OffsetCommitKey] &&
+              coordinatorRecord.value().message().isInstanceOf[OffsetCommitValue])
+            .exists(coordinatorRecord => {
+              val offsetCommitKey = coordinatorRecord.key().message().asInstanceOf[OffsetCommitKey]
+              val offsetCommitValue = coordinatorRecord.value().message().asInstanceOf[OffsetCommitValue]
+              offsetCommitKey.group() == groupId &&
+                offsetCommitKey.topic() == topic &&
+                offsetCommitKey.partition == partition &&
+                offsetCommitValue.offset() == 100 + version
+            })
+        }, "Txn offset commit not found")


This is a bit extreme. I would rather prefer to use fetchOffset to validate the committed offsets.

Thanks for the review, I have addressed all comments in the latest commits. Also loop the integration test 247 times and all passed.

I=0; while ./gradlew cleanTest core:test --tests kafka.server.TxnOffsetCommitRequestTest --rerun --fail-fast; do (( I=$I+1 )); echo "Completed run: $I"; sleep 1; done

dajac

@brandboat Thanks for the update and your patience. I left a few more minor suggestions.

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

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

dajac · 2024-11-29T12:57:38Z

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

+      // Verify that the TXN_OFFSET_COMMIT request is processed correctly when the member ID
+      // and generation ID are known. This validation starts from version 3, as the member ID
+      // must not be empty from version 3 onwards.
+      if (version >= 3) {
+        verifyTxnCommitAndFetch(
+          groupId = groupId,
+          memberId = memberId,
+          generationId = memberEpoch,
+          offset = 200 + version,
+          version = version.toShort
+        )
+      }


Would it be possible to add a negative cases too? We could add them in this test or in another one.

Perhaps we can put the negative case to TxnOffsetCommitRequestTest.java ? I thought the TxnOffsetCommitRequestTest should take care the scenario that GroupMetadata is defined while version < 3.
Let me know if you have any concern. Thanks.

By negative cases, I meant a request with invalid memberId and/or generationId to ensure that the request is rejected with the correct error. Is it worth it?

Agree, I added those test in the latest commit, thanks for the suggestion !

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

dajac

@brandboat Thanks for the update and your patience. We're almost there! I replied to one of the open comments and left a small nit.

dajac · 2024-12-02T13:09:04Z

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

+      // Verify that the TXN_OFFSET_COMMIT request is processed correctly when the member ID
+      // and generation ID are known. This validation starts from version 3, as the member ID
+      // must not be empty from version 3 onwards.
+      if (version >= 3) {
+        verifyTxnCommitAndFetch(
+          groupId = groupId,
+          memberId = memberId,
+          generationId = memberEpoch,
+          offset = 200 + version,
+          version = version.toShort
+        )
+      }


By negative cases, I meant a request with invalid memberId and/or generationId to ensure that the request is rejected with the correct error. Is it worth it?

dajac · 2024-12-02T13:10:50Z

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

+
+    createTopic(topic, 1)
+
+    def verifyTxnCommitAndFetch(


nit: This method is pretty long to be inlined now. Would it make sense to extract it?

chia7712

@brandboat thanks for this great patch. some trivial comments are left. PTAL

chia7712 · 2024-12-02T17:58:01Z

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

+    expectedTxnCommitError: Errors
+  ): Unit = {
+    var producerIdAndEpoch: ProducerIdAndEpoch = null
+    // Wait until ALLOCATE_PRODUCER_ID request finished


It seems that a more accurate comment would be "Wait until the coordinator finishes loading."

chia7712 · 2024-12-02T18:00:33Z

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

+        val topicRecord = groupIdRecord.topics.asScala.find(_.name == topic).head
+        val partitionRecord = topicRecord.partitions.asScala.find(_.partitionIndex == partition).head
+        assertEquals(offset, partitionRecord.committedOffset)
+        true


offset == partitionRecord.committedOffset

chia7712 · 2024-12-02T18:17:42Z

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

+      expectedError = Errors.NONE
+    )
+
+    if (expectedTxnCommitError != Errors.NONE) return


Could you please confirm that committed offsets remain unchanged after a failed TXN_OFFSET_COMMIT?

Fixed in the latest commit, thanks.

dajac

@brandboat Thanks for the update. Overall, LGTM. I left a few nits.

dajac · 2024-12-03T08:28:11Z

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

+
+    for (version <- 0 to ApiKeys.TXN_OFFSET_COMMIT.latestVersion(isUnstableApiEnabled)) {
+      // Verify that the TXN_OFFSET_COMMIT request is processed correctly when member id is UNKNOWN_MEMBER_ID
+      // and generation id is UNKNOWN_GENERATION_ID under all api versions


nit: . at the end of the comment.

dajac · 2024-12-03T08:28:22Z

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

+          expectedTxnCommitError = Errors.NONE
+        )
+
+        // Verify TXN_OFFSET_COMMIT request failed with incorrect memberId


dajac · 2024-12-03T08:28:27Z

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

+          expectedTxnCommitError = Errors.UNKNOWN_MEMBER_ID
+        )
+
+        // Verify TXN_OFFSET_COMMIT request failed with incorrect generationId


dajac

LGTM, thanks.

dajac · 2024-12-03T15:25:33Z

@chia7712 Do you want to take another look at it? Otherwise, I can merge it tomorrow.

…th empty member id when using CONSUMER group (apache#17914) There are two issues in KAFKA-18060: 1) New coordinator can't handle the TxnOffsetCommitRequest with empty member id, and TxnOffsetCommitRequest v0-v2 do definitely has an empty member ID, causing ConsumerGroup#validateOffsetCommit to throw an UnknownMemberIdException. This prevents the old producer from calling sendOffsetsToTransaction. Note that TxnOffsetCommitRequest versions v0-v2 are included in KIP-896, so it seems the new coordinator should support that operations 2) The deprecated API Producer#sendOffsetsToTransaction does not use v0-v2 to send TxnOffsetCommitRequest with an empty member ID. Unfortunately, it has been released for a while. Therefore, the new coordinator needs to handle TxnOffsetCommitRequest with an empty member ID for all versions. Taken from the two issues above, we need to handle empty member id in all API versions when new coordinator are dealing with TxnOffsetCommitRequest. Reviewers: David Jacot <[email protected]>, Chia-Ping Tsai <[email protected]>

KAFKA-18060 new coordinator does not handle TxnOffsetCommitRequest wi…

46bc303

…th empty member id when using CONSUMER group

github-actions bot added the core Kafka Broker label Nov 22, 2024

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

dajac self-requested a review November 22, 2024 15:52

dajac changed the title ~~KAFKA-18060 new coordinator does not handle TxnOffsetCommitRequest with empty member id when using CONSUMER group~~ KAFKA-18060: new coordinator does not handle TxnOffsetCommitRequest with empty member id when using CONSUMER group Nov 22, 2024

dajac reviewed Nov 22, 2024

View reviewed changes

brandboat added 2 commits November 23, 2024 18:41

Address comments

54dfc6c

Merge remote-tracking branch 'origin/trunk' into KAFKA-18060

4845ffa

dajac reviewed Nov 26, 2024

View reviewed changes

brandboat added 2 commits November 28, 2024 21:46

Address comments

516038c

Merge remote-tracking branch 'origin/trunk' into KAFKA-18060

d0575fc

dajac reviewed Nov 29, 2024

View reviewed changes

brandboat added 2 commits November 30, 2024 12:42

Address comments

a5162f4

Merge remote-tracking branch 'origin/trunk' into KAFKA-18060

bb6e1c1

github-actions bot added the clients label Nov 30, 2024

dajac reviewed Dec 2, 2024

View reviewed changes

brandboat added 4 commits December 3, 2024 00:54

Address comments

e617b22

Merge remote-tracking branch 'origin/trunk' into KAFKA-18060

67231ae

Add private modifier

50728f8

Add condition to avoid misunderstanding

037dd30

chia7712 reviewed Dec 2, 2024

View reviewed changes

Address comments

81bd0b9

dajac approved these changes Dec 3, 2024

View reviewed changes

brandboat added 2 commits December 3, 2024 16:45

Address comment

7563ee4

Merge remote-tracking branch 'origin/trunk' into KAFKA-18060

ff1d84b

dajac approved these changes Dec 3, 2024

View reviewed changes

chia7712 approved these changes Dec 3, 2024

View reviewed changes

chia7712 merged commit ac8b3df into apache:trunk Dec 3, 2024
8 checks passed

brandboat deleted the KAFKA-18060 branch December 4, 2024 00:55

		@@ -26,6 +26,7 @@
		import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;

		adminClient = cluster.admin()
		adminClient.createTopics(singleton(new NewTopic(topic, 1, 1.toShort)))

		memberId = JoinGroupRequest.UNKNOWN_MEMBER_ID,
		generationId = JoinGroupRequest.UNKNOWN_GENERATION_ID,

KAFKA-18060: new coordinator does not handle TxnOffsetCommitRequest with empty member id when using CONSUMER group #17914

KAFKA-18060: new coordinator does not handle TxnOffsetCommitRequest with empty member id when using CONSUMER group #17914

Conversation

brandboat commented Nov 22, 2024 • edited Loading

Committer Checklist (excluded from commit message)

dajac commented Nov 22, 2024

dajac 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

dajac 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

dajac 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

brandboat Dec 2, 2024 • edited Loading

Choose a reason for hiding this comment

dajac left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 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

dajac 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

dajac left a comment

Choose a reason for hiding this comment

dajac commented Dec 3, 2024

brandboat commented Nov 22, 2024 •

edited

Loading

brandboat Dec 2, 2024 •

edited

Loading