KAFKA-18914 Migrate ConsumerRebootstrapTest to use new test infra (#19154)

Migrate ConsumerRebootstrapTest to the new test infra and remove the old
Scala test.

The PR changed three things.
* Migrated `ConsumerRebootstrapTest` to new test infra and removed the
old Scala test.
* Updated the original test case to cover rebootstrap scenarios.
* Integrated `ConsumerRebootstrapTest` into `ClientRebootstrapTest` in
the `client-integration-tests` module.
* Removed the `RebootstrapTest.scala`.

Default `ConsumerRebootstrap` config:
> properties.put(CommonClientConfigs.METADATA_RECOVERY_STRATEGY_CONFIG,
"rebootstrap");

properties.put(CommonClientConfigs.METADATA_RECOVERY_REBOOTSTRAP_TRIGGER_MS_CONFIG,
"300000");

properties.put(CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_CONFIG,
"10000");

properties.put(CommonClientConfigs.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_CONFIG,
"30000");
properties.put(CommonClientConfigs.RECONNECT_BACKOFF_MS_CONFIG, "50L");
properties.put(CommonClientConfigs.RECONNECT_BACKOFF_MAX_MS_CONFIG,
"1000L");

The test case for the consumer with enabled rebootstrap
![Screenshot 2025-03-22 at 9 48
13 PM](https://github.com/user-attachments/assets/8470549f-a24c-43fa-ae44-789cbf422a63)


The test case for the consumer with disabled rebootstrap
![Screenshot 2025-03-22 at 9 47
22 PM](https://github.com/user-attachments/assets/0a183464-6a74-449f-8e71-d641a6ea5bb1)

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

Author: ClarkChen

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/ClientRebootstrapTest.java

@@ -17,9 +17,14 @@
 package org.apache.kafka.clients;
 
 import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.GroupProtocol;
+import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.config.TopicConfig;
 import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.TestUtils;
 import org.apache.kafka.common.test.api.ClusterConfigProperty;
 import org.apache.kafka.common.test.api.ClusterTest;
 import org.apache.kafka.common.test.api.Type;
@@ -38,6 +43,7 @@
 
 public class ClientRebootstrapTest {
     private static final String TOPIC = "topic";
+    private static final int PARTITIONS = 1;
     private static final int REPLICAS = 2;
 
     @ClusterTest(
@@ -55,7 +61,7 @@ public void testAdminRebootstrap(ClusterInstance clusterInstance) {
         clusterInstance.shutdownBroker(broker0);
 
         try (var admin = clusterInstance.admin()) {
-            admin.createTopics(List.of(new NewTopic(TOPIC, 1, (short) REPLICAS)));
+            admin.createTopics(List.of(new NewTopic(TOPIC, PARTITIONS, (short) REPLICAS)));
 
             // Only the broker 1 is available for the admin client during the bootstrap.
             assertDoesNotThrow(() -> admin.listTopics().names().get(timeout, TimeUnit.SECONDS).contains(TOPIC));
@@ -84,7 +90,7 @@ public void testAdminRebootstrapDisabled(ClusterInstance clusterInstance) {
         clusterInstance.shutdownBroker(broker0);
 
         var admin = clusterInstance.admin(Map.of(CommonClientConfigs.METADATA_RECOVERY_STRATEGY_CONFIG, "none"));
-        admin.createTopics(List.of(new NewTopic(TOPIC, 1, (short) REPLICAS)));
+        admin.createTopics(List.of(new NewTopic(TOPIC, PARTITIONS, (short) REPLICAS)));
 
         // Only the broker 1 is available for the admin client during the bootstrap.
         assertDoesNotThrow(() -> admin.listTopics().names().get(60, TimeUnit.SECONDS).contains(TOPIC));
@@ -109,7 +115,7 @@ public void testAdminRebootstrapDisabled(ClusterInstance clusterInstance) {
     )
     public void testProducerRebootstrap(ClusterInstance clusterInstance) throws ExecutionException, InterruptedException {
         try (var admin = clusterInstance.admin()) {
-            admin.createTopics(List.of(new NewTopic(TOPIC, 1, (short) REPLICAS)));
+            admin.createTopics(List.of(new NewTopic(TOPIC, PARTITIONS, (short) REPLICAS)));
         }
 
         var broker0 = 0;
@@ -144,7 +150,7 @@ public void testProducerRebootstrap(ClusterInstance clusterInstance) throws Exec
     )
     public void testProducerRebootstrapDisabled(ClusterInstance clusterInstance) throws ExecutionException, InterruptedException {
         try (var admin = clusterInstance.admin()) {
-            admin.createTopics(List.of(new NewTopic(TOPIC, 1, (short) REPLICAS)));
+            admin.createTopics(List.of(new NewTopic(TOPIC, PARTITIONS, (short) REPLICAS)));
         }
 
         var broker0 = 0;
@@ -168,4 +174,123 @@ public void testProducerRebootstrapDisabled(ClusterInstance clusterInstance) thr
         // Since the brokers cached during the bootstrap are offline, the producer needs to wait the default timeout for other threads.
         producer.close(Duration.ZERO);
     }
+
+    public void consumerRebootstrap(ClusterInstance clusterInstance, GroupProtocol groupProtocol) throws InterruptedException, ExecutionException {
+        clusterInstance.createTopic(TOPIC, PARTITIONS, (short) REPLICAS);
+
+        var broker0 = 0;
+        var broker1 = 1;
+        var partitions = List.of(new TopicPartition(TOPIC, 0));
+
+        try (var producer = clusterInstance.producer(Map.of(ProducerConfig.ACKS_CONFIG, "-1"))) {
+            var recordMetadata = producer.send(new ProducerRecord<>(TOPIC, "value 0".getBytes())).get();
+            assertEquals(0, recordMetadata.offset());
+        }
+
+        clusterInstance.shutdownBroker(broker0);
+
+        try (var consumer = clusterInstance.consumer(Map.of(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name))) {
+            // Only the server 1 is available for the consumer during the bootstrap.
+            consumer.assign(partitions);
+            consumer.seekToBeginning(partitions);
+            TestUtils.waitForCondition(() -> consumer.poll(Duration.ofMillis(100)).count() == 1, 10 * 1000, "Failed to poll data.");
+
+            // Bring back the server 0 and shut down 1.
+            clusterInstance.shutdownBroker(broker1);
+            clusterInstance.startBroker(broker0);
+
+            try (var producer = clusterInstance.producer(Map.of(ProducerConfig.ACKS_CONFIG, "-1"))) {
+                var recordMetadata = producer.send(new ProducerRecord<>(TOPIC, "value 1".getBytes())).get();
+                assertEquals(1, recordMetadata.offset());
+            }
+
+            // The server 1 originally cached during the bootstrap, is offline.
+            // However, the server 0 from the bootstrap list is online.
+            TestUtils.waitForCondition(() -> consumer.poll(Duration.ofMillis(100)).count() == 1, 10 * 1000, "Failed to poll data.");
+        }
+    }
+
+    @ClusterTest(
+        brokers = REPLICAS,
+        types = {Type.KRAFT},
+        serverProperties = {
+            @ClusterConfigProperty(key = TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+            @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2"),
+        })
+    public void testClassicConsumerRebootstrap(ClusterInstance clusterInstance) throws InterruptedException, ExecutionException {
+        consumerRebootstrap(clusterInstance, GroupProtocol.CLASSIC);
+    }
+
+    @ClusterTest(
+        brokers = REPLICAS,
+        types = {Type.KRAFT},
+        serverProperties = {
+            @ClusterConfigProperty(key = TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+            @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2"),
+        })
+    public void testConsumerRebootstrap(ClusterInstance clusterInstance) throws InterruptedException, ExecutionException {
+        consumerRebootstrap(clusterInstance, GroupProtocol.CONSUMER);
+    }
+
+    public void consumerRebootstrapDisabled(ClusterInstance clusterInstance, GroupProtocol groupProtocol) throws InterruptedException, ExecutionException {
+        clusterInstance.createTopic(TOPIC, PARTITIONS, (short) REPLICAS);
+
+        var broker0 = 0;
+        var broker1 = 1;
+        var tp = new TopicPartition(TOPIC, 0);
+
+        try (var producer = clusterInstance.producer(Map.of(ProducerConfig.ACKS_CONFIG, "-1"))) {
+            var recordMetadata = producer.send(new ProducerRecord<>(TOPIC, "value 0".getBytes())).get();
+            assertEquals(0, recordMetadata.offset());
+        }
+
+        clusterInstance.shutdownBroker(broker0);
+
+        try (var consumer = clusterInstance.consumer(Map.of(
+            CommonClientConfigs.METADATA_RECOVERY_STRATEGY_CONFIG, "none",
+            ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name)
+        )) {
+            // Only the server 1 is available for the consumer during the bootstrap.
+            consumer.assign(List.of(tp));
+            consumer.seekToBeginning(List.of(tp));
+            TestUtils.waitForCondition(() -> consumer.poll(Duration.ofMillis(100)).count() == 1, 10 * 1000, "Failed to poll data.");
+
+            // Bring back the server 0 and shut down 1.
+            clusterInstance.shutdownBroker(broker1);
+            clusterInstance.startBroker(broker0);
+
+            try (var producer = clusterInstance.producer(Map.of(ProducerConfig.ACKS_CONFIG, "-1"))) {
+                var recordMetadata = producer.send(new ProducerRecord<>(TOPIC, "value 1".getBytes())).get();
+                assertEquals(1, recordMetadata.offset());
+            }
+
+            // The server 1 originally cached during the bootstrap, is offline.
+            // However, the server 0 from the bootstrap list is online.
+            assertEquals(0, consumer.poll(Duration.ofMillis(100)).count());
+        }
+    }
+
+    @ClusterTest(
+        brokers = REPLICAS,
+        types = {Type.KRAFT},
+        serverProperties = {
+            @ClusterConfigProperty(key = TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+            @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2")
+        }
+    )
+    public void testClassicConsumerRebootstrapDisabled(ClusterInstance clusterInstance) throws InterruptedException, ExecutionException {
+        consumerRebootstrapDisabled(clusterInstance, GroupProtocol.CLASSIC);
+    }
+
+    @ClusterTest(
+        brokers = REPLICAS,
+        types = {Type.KRAFT},
+        serverProperties = {
+            @ClusterConfigProperty(key = TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+            @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2")
+        }
+    )
+    public void testConsumerRebootstrapDisabled(ClusterInstance clusterInstance) throws InterruptedException, ExecutionException {
+        consumerRebootstrapDisabled(clusterInstance, GroupProtocol.CONSUMER);
+    }
 }