MINOR: Use Producer interface and ClusterInstance producer factory (#18197)

Reviewers: David Arthur <[email protected]>, Chia-Ping Tsai <[email protected]>

Author: peterxcli

core/src/test/java/kafka/admin/AdminFenceProducersTest.java

@@ -20,14 +20,13 @@
 import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.FenceProducersOptions;
-import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.errors.ApiException;
 import org.apache.kafka.common.errors.InvalidProducerEpochException;
 import org.apache.kafka.common.errors.ProducerFencedException;
 import org.apache.kafka.common.errors.TimeoutException;
-import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.apache.kafka.common.test.api.ClusterConfigProperty;
 import org.apache.kafka.common.test.api.ClusterInstance;
 import org.apache.kafka.common.test.api.ClusterTest;
@@ -42,7 +41,6 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.Properties;
 import java.util.concurrent.ExecutionException;
 
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -68,22 +66,16 @@ public class AdminFenceProducersTest {
         this.clusterInstance = clusterInstance;
     }
 
-    private KafkaProducer<byte[], byte[]> createProducer() {
-        Properties config = new Properties();
-        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers());
-        config.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, TXN_ID);
-        config.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "2000");
-        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
-        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
-
-        return new KafkaProducer<>(config);
+    private Producer<byte[], byte[]> createProducer() {
+        return clusterInstance.producer(Map.of(ProducerConfig.TRANSACTIONAL_ID_CONFIG, TXN_ID,
+                ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "2000"));
     }
 
     @ClusterTest
     void testFenceAfterProducerCommit() throws Exception {
         clusterInstance.createTopic(TOPIC_NAME, 1, (short) 1);
 
-        try (KafkaProducer<byte[], byte[]> producer = createProducer();
+        try (Producer<byte[], byte[]> producer = createProducer();
              Admin adminClient = clusterInstance.admin()) {
             producer.initTransactions();
             producer.beginTransaction();
@@ -125,7 +117,7 @@ void testFenceProducerTimeoutMs() {
     void testFenceBeforeProducerCommit() throws Exception {
         clusterInstance.createTopic(TOPIC_NAME, 1, (short) 1);
 
-        try (KafkaProducer<byte[], byte[]> producer = createProducer();
+        try (Producer<byte[], byte[]> producer = createProducer();
              Admin adminClient = clusterInstance.admin()) {
 
             producer.initTransactions();

tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteOffsetsConsumerGroupCommandIntegrationTest.java

@@ -25,13 +25,12 @@
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.GroupProtocol;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
 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.protocol.Errors;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
-import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.apache.kafka.common.test.api.ClusterConfig;
 import org.apache.kafka.common.test.api.ClusterInstance;
 import org.apache.kafka.common.test.api.ClusterTemplate;
@@ -47,7 +46,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.Properties;
 
 import static org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
@@ -226,19 +224,13 @@ private void testWithConsumerGroup(String inputTopic,
     }
 
     private void produceRecord(String topic) {
-        try (KafkaProducer<byte[], byte[]> producer = createProducer()) {
+        try (Producer<byte[], byte[]> producer = createProducer()) {
             assertDoesNotThrow(() -> producer.send(new ProducerRecord<>(topic, 0, null, null)).get());
         }
     }
 
-    private KafkaProducer<byte[], byte[]> createProducer() {
-        Properties config = new Properties();
-        config.putIfAbsent(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers());
-        config.putIfAbsent(ProducerConfig.ACKS_CONFIG, "-1");
-        config.putIfAbsent(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
-        config.putIfAbsent(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
-
-        return new KafkaProducer<>(config);
+    private Producer<byte[], byte[]> createProducer() {
+        return clusterInstance.producer(Map.of(ProducerConfig.ACKS_CONFIG, "-1"));
     }
 
     private Consumer<byte[], byte[]> createConsumer(String group, GroupProtocol groupProtocol) {