KAFKA-18276 Migrate ProducerRebootstrapTest to new test infra (#19046)

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

Default `ProducerRebootstrap` 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 producer with enabled rebootstrap
<img width="1549" alt="Screenshot 2025-03-17 at 10 46 03 PM"
src="https://github.com/user-attachments/assets/547840a6-d79d-4db4-98c0-9b05ed04cf60"
/>

The test case for the producer with disabled rebootstrap
<img width="1552" alt="Screenshot 2025-03-17 at 10 46 47 PM"
src="https://github.com/user-attachments/assets/2248e809-d9d5-4f3b-a24f-ba1aa0fef728"
/>

Reviewers: TengYao Chi <[email protected]>, Ken Huang
<[email protected]>, Chia-Ping Tsai <[email protected]>

Author: clarkwtc

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

@@ -17,6 +17,8 @@
 package org.apache.kafka.clients;
 
 import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.config.TopicConfig;
 import org.apache.kafka.common.test.ClusterInstance;
 import org.apache.kafka.common.test.api.ClusterConfigProperty;
 import org.apache.kafka.common.test.api.ClusterTest;
@@ -26,10 +28,12 @@
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
 public class ClientRebootstrapTest {
@@ -94,4 +98,74 @@ public void testAdminRebootstrapDisabled(ClusterInstance clusterInstance) {
         // Since the brokers cached during the bootstrap are offline, the admin client needs to wait the default timeout for other threads.
         admin.close(Duration.ZERO);
     }
+
+    @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 testProducerRebootstrap(ClusterInstance clusterInstance) throws ExecutionException, InterruptedException {
+        try (var admin = clusterInstance.admin()) {
+            admin.createTopics(List.of(new NewTopic(TOPIC, 1, (short) REPLICAS)));
+        }
+
+        var broker0 = 0;
+        var broker1 = 1;
+
+        // It's ok to shut the leader down, cause the reelection is small enough to the producer timeout.
+        clusterInstance.shutdownBroker(broker0);
+
+        try (var producer = clusterInstance.producer()) {
+            // Only the broker 1 is available for the producer during the bootstrap.
+            var recordMetadata0 = producer.send(new ProducerRecord<>(TOPIC, "value 0".getBytes())).get();
+            assertEquals(0, recordMetadata0.offset());
+
+            clusterInstance.shutdownBroker(broker1);
+            clusterInstance.startBroker(broker0);
+
+            // Current broker 1 is offline.
+            // However, the broker 0 from the bootstrap list is online.
+            // Should be able to produce records.
+            var recordMetadata1 = producer.send(new ProducerRecord<>(TOPIC, "value 1".getBytes())).get();
+            assertEquals(0, recordMetadata1.offset());
+        }
+    }
+
+    @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 testProducerRebootstrapDisabled(ClusterInstance clusterInstance) throws ExecutionException, InterruptedException {
+        try (var admin = clusterInstance.admin()) {
+            admin.createTopics(List.of(new NewTopic(TOPIC, 1, (short) REPLICAS)));
+        }
+
+        var broker0 = 0;
+        var broker1 = 1;
+
+        // It's ok to shut the leader down, cause the reelection is small enough to the producer timeout.
+        clusterInstance.shutdownBroker(broker0);
+
+        var producer = clusterInstance.producer(Map.of(CommonClientConfigs.METADATA_RECOVERY_STRATEGY_CONFIG, "none"));
+
+        // Only the broker 1 is available for the producer during the bootstrap.
+        var recordMetadata0 = producer.send(new ProducerRecord<>(TOPIC, "value 0".getBytes())).get();
+        assertEquals(0, recordMetadata0.offset());
+
+        clusterInstance.shutdownBroker(broker1);
+        clusterInstance.startBroker(broker0);
+
+        // The broker 1, originally cached during the bootstrap, is offline.
+        // As a result, the producer will throw a TimeoutException when trying to send a message.
+        assertThrows(TimeoutException.class, () -> producer.send(new ProducerRecord<>(TOPIC, "value 1".getBytes())).get(5, TimeUnit.SECONDS));
+        // Since the brokers cached during the bootstrap are offline, the producer needs to wait the default timeout for other threads.
+        producer.close(Duration.ZERO);
+    }
 }