Re: [PR] KAFKA-19141; Persist topic id in OffsetCommit record [kafka]

[via GitHub]

squah-confluent commented on code in PR #19683:
URL: https://github.com/apache/kafka/pull/19683#discussion_r2086490152


##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetAndMetadataTest.java:
##########
@@ -16,52 +16,69 @@
  */
 package org.apache.kafka.coordinator.group;
 
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.message.OffsetCommitRequestData;
 import org.apache.kafka.common.message.TxnOffsetCommitRequestData;
 import org.apache.kafka.coordinator.group.generated.OffsetCommitValue;
 import org.apache.kafka.server.util.MockTime;
 
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
 
 import java.util.OptionalInt;
 import java.util.OptionalLong;
+import java.util.stream.Stream;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 public class OffsetAndMetadataTest {
     @Test
     public void testAttributes() {
+        Uuid topicId = Uuid.randomUuid();
         OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(
             100L,
             OptionalInt.of(10),
             "metadata",
             1234L,
-            OptionalLong.of(5678L)
+            OptionalLong.of(5678L),
+            topicId
         );
 
         assertEquals(100L, offsetAndMetadata.committedOffset);
         assertEquals(OptionalInt.of(10), offsetAndMetadata.leaderEpoch);
         assertEquals("metadata", offsetAndMetadata.metadata);
         assertEquals(1234L, offsetAndMetadata.commitTimestampMs);
         assertEquals(OptionalLong.of(5678L), 
offsetAndMetadata.expireTimestampMs);
+        assertEquals(topicId, offsetAndMetadata.topicId);
     }
 
-    @Test
-    public void testFromRecord() {
+    private static Stream<Uuid> uuids() {
+        return Stream.of(
+            Uuid.ZERO_UUID,
+            Uuid.randomUuid()
+        );
+    }
+
+    @ParameterizedTest
+    @MethodSource("uuids")
+    public void testFromRecord(Uuid uuid) {
         OffsetCommitValue record = new OffsetCommitValue()
             .setOffset(100L)
             .setLeaderEpoch(-1)
             .setMetadata("metadata")
             .setCommitTimestamp(1234L)
-            .setExpireTimestamp(-1L);
+            .setExpireTimestamp(-1L)
+            .setTopicId(uuid);

Review Comment:
   Just to confirm: when we read an older `OffsetCommitValue`, will we get a 
`ZERO_UUID` instead of a `null`?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org