fix message test

Author: TaiJuWu

clients/src/test/java/org/apache/kafka/common/message/MessageTest.java

@@ -184,39 +184,72 @@ public void testJoinGroupRequestVersions() throws Exception {
 
     @Test
     public void testListOffsetsRequestVersions() throws Exception {
-        List<ListOffsetsTopic> v = Collections.singletonList(new ListOffsetsTopic()
-                .setName("topic")
-                .setPartitions(Collections.singletonList(new ListOffsetsPartition()
-                        .setPartitionIndex(0)
-                        .setTimestamp(123L))));
-        Supplier<ListOffsetsRequestData> newRequest = () -> new ListOffsetsRequestData()
-                .setTopics(v)
-                .setReplicaId(0);
-        testAllMessageRoundTrips(newRequest.get());
-        testAllMessageRoundTripsFromVersion((short) 2, newRequest.get().setIsolationLevel(IsolationLevel.READ_COMMITTED.id()));
+        // Test each version with appropriate fields
+        for (short version = ApiKeys.LIST_OFFSETS.oldestVersion(); version <= ApiKeys.LIST_OFFSETS.latestVersion(); ++version) {
+            ListOffsetsRequestData request = new ListOffsetsRequestData()
+                    .setReplicaId(0);
+
+            if (version >= 2) {
+                request.setIsolationLevel(IsolationLevel.READ_COMMITTED.id());
+            }
+
+            if (version >= 12) {
+                // Version 12+: use topic IDs only
+                Uuid topicId = Uuid.randomUuid();
+                List<ListOffsetsTopic> topicsWithIds = Collections.singletonList(new ListOffsetsTopic()
+                        .setTopicId(topicId)
+                        .setPartitions(Collections.singletonList(new ListOffsetsPartition()
+                                .setPartitionIndex(0)
+                                .setTimestamp(123L))));
+                request.setTopics(topicsWithIds);
+            } else {
+                // Version < 12: use topic names only
+                List<ListOffsetsTopic> topicsWithNames = Collections.singletonList(new ListOffsetsTopic()
+                        .setName("topic")
+                        .setPartitions(Collections.singletonList(new ListOffsetsPartition()
+                                .setPartitionIndex(0)
+                                .setTimestamp(123L))));
+                request.setTopics(topicsWithNames);
+            }
+
+            testEquivalentMessageRoundTrip(version, request);
+        }
     }
 
     @Test
     public void testListOffsetsResponseVersions() throws Exception {
-        ListOffsetsPartitionResponse partition = new ListOffsetsPartitionResponse()
-                .setErrorCode(Errors.NONE.code())
-                .setPartitionIndex(0);
-        List<ListOffsetsTopicResponse> topics = Collections.singletonList(new ListOffsetsTopicResponse()
-                .setName("topic")
-                .setPartitions(Collections.singletonList(partition)));
-        Supplier<ListOffsetsResponseData> response = () -> new ListOffsetsResponseData()
-                .setTopics(topics);
         for (short version = ApiKeys.LIST_OFFSETS.oldestVersion(); version <= ApiKeys.LIST_OFFSETS.latestVersion(); ++version) {
-            ListOffsetsResponseData responseData = response.get();
-            responseData.topics().get(0).partitions().get(0)
-                .setOffset(456L)
-                .setTimestamp(123L);
-            if (version > 1) {
-                responseData.setThrottleTimeMs(1000);
-            }
+            ListOffsetsPartitionResponse partition = new ListOffsetsPartitionResponse()
+                    .setErrorCode(Errors.NONE.code())
+                    .setPartitionIndex(0)
+                    .setOffset(456L)
+                    .setTimestamp(123L);
+
             if (version > 3) {
                 partition.setLeaderEpoch(1);
             }
+
+            List<ListOffsetsTopicResponse> topics;
+            if (version >= 12) {
+                // Version 12+: use topic IDs only
+                Uuid topicId = Uuid.randomUuid();
+                topics = Collections.singletonList(new ListOffsetsTopicResponse()
+                        .setTopicId(topicId)
+                        .setPartitions(Collections.singletonList(partition)));
+            } else {
+                // Version < 12: use topic names only
+                topics = Collections.singletonList(new ListOffsetsTopicResponse()
+                        .setName("topic")
+                        .setPartitions(Collections.singletonList(partition)));
+            }
+
+            ListOffsetsResponseData responseData = new ListOffsetsResponseData()
+                    .setTopics(topics);
+
+            if (version > 1) {
+                responseData.setThrottleTimeMs(1000);
+            }
+
             testEquivalentMessageRoundTrip(version, responseData);
         }
     }