MINOR: Add topicId parameter to OffsetAndMetadata.fromRequest for TxnOffsetCommit (#22159)

This patch adds a `topicId` parameter to the
`TxnOffsetCommitRequestPartition` overload of
`OffsetAndMetadata.fromRequest`, mirroring the existing
`OffsetCommitRequestPartition` overload. The single caller in
`OffsetMetadataManager.commitTransactionalOffset` passes
`Uuid.ZERO_UUID` to preserve the current behavior.

The `testFromTransactionalRequest` test is converted to a parameterized
test that covers both `Uuid.ZERO_UUID` and a random topic ID.

Reviewers: Andrew Schofield <aschofield@confluent.io>

Author: dajac

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

@@ -190,6 +190,7 @@ public static OffsetAndMetadata fromRequest(
      * @return An OffsetAndMetadata created from an OffsetCommitRequestPartition request.
      */
     public static OffsetAndMetadata fromRequest(
+        Uuid topicId,
         TxnOffsetCommitRequestData.TxnOffsetCommitRequestPartition partition,
         long currentTimeMs
     ) {
@@ -200,7 +201,7 @@ public static OffsetAndMetadata fromRequest(
                 OffsetAndMetadata.NO_METADATA : partition.committedMetadata(),
             currentTimeMs,
             OptionalLong.empty(),
-            Uuid.ZERO_UUID
+            topicId
         );
     }
 }

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

@@ -737,6 +737,7 @@ public CoordinatorResult<TxnOffsetCommitResponseData, CoordinatorRecord> commitT
                         .setErrorCode(Errors.NONE.code()));
 
                     final OffsetAndMetadata offsetAndMetadata = OffsetAndMetadata.fromRequest(
+                        Uuid.ZERO_UUID,
                         partition,
                         currentTimeMs
                     );

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetAndMetadataTest.java

@@ -161,8 +161,9 @@ public void testFromRequest(Uuid uuid) {
         );
     }
 
-    @Test
-    public void testFromTransactionalRequest() {
+    @ParameterizedTest
+    @MethodSource("uuids")
+    public void testFromTransactionalRequest(Uuid uuid) {
         MockTime time = new MockTime();
 
         TxnOffsetCommitRequestData.TxnOffsetCommitRequestPartition partition =
@@ -179,8 +180,9 @@ public void testFromTransactionalRequest() {
                 "",
                 time.milliseconds(),
                 OptionalLong.empty(),
-                Uuid.ZERO_UUID
+                uuid
             ), OffsetAndMetadata.fromRequest(
+                uuid,
                 partition,
                 time.milliseconds()
             )
@@ -197,8 +199,9 @@ public void testFromTransactionalRequest() {
                 "hello",
                 time.milliseconds(),
                 OptionalLong.empty(),
-                Uuid.ZERO_UUID
+                uuid
             ), OffsetAndMetadata.fromRequest(
+                uuid,
                 partition,
                 time.milliseconds()
             )