KAFKA-17689 Migrate ListOffsetsTest to new test infra (#22369)

Migrate `ListOffsetsTest` to the new test infrastructure

Reviewers: Ken Huang <s7133700@gmail.com>, Chia-Ping Tsai
 <chia7712@gmail.com>

Author: unknowntpo

storage/src/test/java/org/apache/kafka/tiered/storage/integration/ListOffsetsTest.java

@@ -17,43 +17,58 @@
 package org.apache.kafka.tiered.storage.integration;
 
 import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.GroupProtocol;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfig;
+import org.apache.kafka.common.test.api.ClusterTemplate;
+import org.apache.kafka.common.test.api.Type;
 import org.apache.kafka.common.utils.MockTime;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.storage.internals.log.EpochEntry;
+import org.apache.kafka.tiered.storage.TieredStorageTestAction;
 import org.apache.kafka.tiered.storage.TieredStorageTestBuilder;
-import org.apache.kafka.tiered.storage.TieredStorageTestHarness;
+import org.apache.kafka.tiered.storage.TieredStorageTestContext;
 import org.apache.kafka.tiered.storage.specs.KeyValueSpec;
 
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.MethodSource;
-
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
+import java.util.Set;
 
 import static org.apache.kafka.common.record.internal.RecordBatch.NO_PARTITION_LEADER_EPOCH;
 import static org.apache.kafka.common.utils.Utils.mkEntry;
 import static org.apache.kafka.common.utils.Utils.mkMap;
 import static org.apache.kafka.server.log.remote.storage.LocalTieredStorageEvent.EventType.DELETE_SEGMENT;
+import static org.apache.kafka.tiered.storage.utils.TieredStorageTestUtils.createServerPropsForRemoteStorage;
+
+public final class ListOffsetsTest {
+
+    private static final int BROKER_COUNT = 3;
+    private static final int NUM_REMOTE_LOG_METADATA_PARTITIONS = 5;
 
-public class ListOffsetsTest extends TieredStorageTestHarness {
-    @Override
-    public int brokerCount() {
-        return 2;
+    private static List<ClusterConfig> clusterConfig() {
+        return List.of(ClusterConfig.defaultBuilder()
+                .setTypes(Set.of(Type.KRAFT))
+                .setBrokers(BROKER_COUNT)
+                .setServerProperties(createServerPropsForRemoteStorage(
+                        ListOffsetsTest.class.getSimpleName().toLowerCase(Locale.ROOT),
+                        BROKER_COUNT,
+                        NUM_REMOTE_LOG_METADATA_PARTITIONS))
+                .build());
     }
 
-    /**
-     * We are running this test only for the Kraft mode, since ZK mode is deprecated now. Note that:
-     * 1. In Kraft mode, the leader-epoch gets bumped only for leader-election (0 -> 1) and not for reassignment.
-     */
-    @ParameterizedTest(name = "{displayName}.groupProtocol={0}")
-    @MethodSource("getTestGroupProtocolParametersAll")
-    @Override
-    public void executeTieredStorageTest(String groupProtocol) {
-        super.executeTieredStorageTest(groupProtocol);
+    @ClusterTemplate("clusterConfig")
+    public void testListOffsetsWithClassicGroupProtocol(ClusterInstance clusterInstance) throws Exception {
+        executeListOffsetsTest(clusterInstance, GroupProtocol.CLASSIC);
     }
 
-    @Override
-    protected void writeTestSpecifications(TieredStorageTestBuilder builder) {
+    @ClusterTemplate("clusterConfig")
+    public void testListOffsetsWithConsumerGroupProtocol(ClusterInstance clusterInstance) throws Exception {
+        executeListOffsetsTest(clusterInstance, GroupProtocol.CONSUMER);
+    }
+
+    private void executeListOffsetsTest(ClusterInstance clusterInstance, GroupProtocol groupProtocol) throws Exception {
         final int broker0 = 0;
         final int broker1 = 1;
         final String topicA = "topicA";
@@ -62,6 +77,7 @@ protected void writeTestSpecifications(TieredStorageTestBuilder builder) {
         final long timestamp = time.milliseconds();
         final Map<Integer, List<Integer>> assignment = mkMap(mkEntry(p0, List.of(broker0, broker1)));
 
+        TieredStorageTestBuilder builder = new TieredStorageTestBuilder();
         builder
                 .createTopic(topicA, 1, 2, 2, assignment, true)
                 // send records to partition 0 and expect the first segment to be offloaded
@@ -120,5 +136,18 @@ protected void writeTestSpecifications(TieredStorageTestBuilder builder) {
                 .expectListOffsets(topicA, p0, OffsetSpec.earliestLocal(), new EpochEntry(1, 4))
                 .expectListOffsets(topicA, p0, OffsetSpec.latestTiered(), new EpochEntry(NO_PARTITION_LEADER_EPOCH, 3))
                 .expectListOffsets(topicA, p0, OffsetSpec.latest(), new EpochEntry(1, 6));
+
+        Map<String, Object> extraConsumerProps = Map.of(
+                ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name().toLowerCase(Locale.ROOT)
+        );
+        try (TieredStorageTestContext context = new TieredStorageTestContext(clusterInstance, extraConsumerProps)) {
+            try {
+                for (TieredStorageTestAction action : builder.complete()) {
+                    action.execute(context);
+                }
+            } finally {
+                context.printReport(System.out);
+            }
+        }
     }
 }