fix some test

Author: TaiJuWu

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

@@ -182,43 +182,47 @@ public void testJoinGroupRequestVersions() throws Exception {
         testAllMessageRoundTripsFromVersion((short) 5, newRequest.get().setGroupInstanceId("instanceId"));
     }
 
-    @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()));
+    @ParameterizedTest
+    @ApiKeyVersionsSource(apiKey = ApiKeys.LIST_OFFSETS)
+    public void testListOffsetsRequestVersions(short version) throws Exception {
+        ListOffsetsRequestData request = new ListOffsetsRequestData()
+            .setReplicaId(0)
+            .setIsolationLevel(version >= 2 ? IsolationLevel.READ_COMMITTED.id() : 0)
+            .setTopics(singletonList(
+                new ListOffsetsTopic()
+                    .setTopicId(version >= 12 ? Uuid.randomUuid() : Uuid.ZERO_UUID)
+                    .setName(version < 12 ? "topic" : "")
+                    .setPartitions(singletonList(
+                        new ListOffsetsPartition()
+                            .setPartitionIndex(0)
+                            .setCurrentLeaderEpoch(version >= 4 ? 10 : -1)
+                            .setTimestamp(123L)
+                    ))
+            ));
+
+        testMessageRoundTrip(version, request, 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);
-            }
-            if (version > 3) {
-                partition.setLeaderEpoch(1);
-            }
-            testEquivalentMessageRoundTrip(version, responseData);
-        }
+    @ParameterizedTest
+    @ApiKeyVersionsSource(apiKey = ApiKeys.LIST_OFFSETS)
+    public void testListOffsetsResponseVersions(short version) throws Exception {
+        ListOffsetsResponseData response = new ListOffsetsResponseData()
+            .setThrottleTimeMs(version >= 2 ? 1000 : 0)
+            .setTopics(singletonList(
+                new ListOffsetsTopicResponse()
+                    .setTopicId(version >= 12 ? Uuid.randomUuid() : Uuid.ZERO_UUID)
+                    .setName(version < 12 ? "topic" : "")
+                    .setPartitions(singletonList(
+                        new ListOffsetsPartitionResponse()
+                            .setPartitionIndex(0)
+                            .setErrorCode(Errors.NONE.code())
+                            .setTimestamp(123L)
+                            .setOffset(456L)
+                            .setLeaderEpoch(version >= 4 ? 1 : -1)
+                    ))
+            ));
+
+        testMessageRoundTrip(version, response, response);
     }
 
     @Test

server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java

@@ -274,7 +274,9 @@ public short fetchRequestVersion() {
     }
 
     public short listOffsetRequestVersion() {
-        if (this.isAtLeast(IBP_4_2_IV1)) {
+        if (this.isAtLeast(IBP_4_3_IV0)) {
+            return 12;
+        } else if (this.isAtLeast(IBP_4_2_IV1)) {
             return 11;
         } else if (this.isAtLeast(IBP_4_0_IV3)) {
             return 10;

storage/src/test/java/org/apache/kafka/server/purgatory/DelayedRemoteListOffsetsTest.java

@@ -193,7 +193,7 @@ public void testResponseOnPartialError() {
         Map<TopicIdPartition, ListOffsetsPartitionStatus> statusByPartition = Map.of(
             new TopicIdPartition(Uuid.randomUuid(), 0, "test"), ListOffsetsPartitionStatus.builder().futureHolderOpt(Optional.of(holder)).build(),
             new TopicIdPartition(Uuid.randomUuid(), 1, "test"), ListOffsetsPartitionStatus.builder().futureHolderOpt(Optional.of(holder)).build(),
-            new TopicIdPartition(Uuid.randomUuid(), 0, "test1"), ListOffsetsPartitionStatus.builder().futureHolderOpt(Optional.of(holder)).build()
+            new TopicIdPartition(Uuid.randomUuid(), 0, "test1"), ListOffsetsPartitionStatus.builder().futureHolderOpt(Optional.of(errorFutureHolder)).build()
         );
 
         DelayedRemoteListOffsets delayedRemoteListOffsets = new DelayedRemoteListOffsets(delayMs, 5, statusByPartition, partitionOrException, responseCallback);