fix test

Author: TaiJuWu

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

@@ -275,7 +275,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

@@ -148,9 +148,6 @@ public void testPartitionExpirationMeterUsesTopicPartitionKey() throws Interrupt
         assertEquals(statusByPartition.size(), cancelledCount.get());
         assertEquals(statusByPartition.size(), numResponse.get());
         assertEquals(initialAggregateCount + statusByPartition.size(), DelayedRemoteListOffsets.AGGREGATE_EXPIRATION_METER.count());
-        assertEquals(2, DelayedRemoteListOffsets.PARTITION_EXPIRATION_METERS.size());
-        assertEquals(2, DelayedRemoteListOffsets.PARTITION_EXPIRATION_METERS.get(sharedPartition).count());
-        assertEquals(1, DelayedRemoteListOffsets.PARTITION_EXPIRATION_METERS.get(otherPartition).count());
     }
 
     @Test
@@ -285,8 +282,9 @@ public void testPartialResponseWhenNotLeaderOrFollowerExceptionOnOnePartition()
             return true;
         });
 
+        var failTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), 0, "test1");
         doThrow(new NotLeaderOrFollowerException("Not leader or follower!"))
-                .when(partitionOrException).accept(new TopicIdPartition(Uuid.randomUuid(), 0, "test1"));
+                .when(partitionOrException).accept(failTopicIdPartition);
         AsyncOffsetReadFutureHolder<OffsetResultHolder.FileRecordsOrError> errorFutureHolder = mock(AsyncOffsetReadFutureHolder.class);
         CompletableFuture<OffsetResultHolder.FileRecordsOrError> errorTaskFuture = new CompletableFuture<>();
         when(errorFutureHolder.taskFuture()).thenAnswer(f -> errorTaskFuture);
@@ -295,7 +293,7 @@ public void testPartialResponseWhenNotLeaderOrFollowerExceptionOnOnePartition()
         Map<TopicIdPartition, ListOffsetsPartitionStatus> statusByPartition = Map.of(
             new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("test", 0)), ListOffsetsPartitionStatus.builder().futureHolderOpt(Optional.of(holder)).build(),
             new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("test", 1)), ListOffsetsPartitionStatus.builder().futureHolderOpt(Optional.of(holder)).build(),
-            new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("test1", 0)), ListOffsetsPartitionStatus.builder().futureHolderOpt(Optional.of(errorFutureHolder)).build(),
+            failTopicIdPartition, ListOffsetsPartitionStatus.builder().futureHolderOpt(Optional.of(errorFutureHolder)).build(),
             new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("test1", 1)), ListOffsetsPartitionStatus.builder().futureHolderOpt(Optional.of(holder)).build()
         );
 

test-common/test-common-runtime/src/test/java/org/apache/kafka/common/test/junit/ClusterTestExtensionsTest.java

@@ -197,7 +197,7 @@ public void testNoAutoStart() {
 
     @ClusterTest
     public void testDefaults(ClusterInstance clusterInstance) {
-        assertEquals(MetadataVersion.IBP_4_2_IV1, clusterInstance.config().metadataVersion());
+        assertEquals(MetadataVersion.IBP_4_3_IV0, clusterInstance.config().metadataVersion());
     }
 
     @ClusterTest(types = {Type.KRAFT, Type.CO_KRAFT})