From 8167da97bf2fc15b6dddee1465f41030cd128ca1 Mon Sep 17 00:00:00 2001 From: Konstantine Karantasis Date: Tue, 22 Mar 2022 15:17:51 -0700 Subject: [PATCH] Revert "KAFKA-7077: Use default producer settings in Connect Worker (#11475)" This reverts commit 76cf7a5793702b55e2cfd98a375f8f1708ff32c3. --- .../src/main/java/org/apache/kafka/connect/runtime/Worker.java | 2 ++ .../test/java/org/apache/kafka/connect/runtime/WorkerTest.java | 2 ++ 2 files changed, 4 insertions(+) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java index 45c7d9f6d13a9..582271a196335 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java @@ -648,6 +648,8 @@ static Map producerConfigs(ConnectorTaskId id, // These settings will execute infinite retries on retriable exceptions. They *may* be overridden via configs passed to the worker, // but this may compromise the delivery guarantees of Kafka Connect. producerProps.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, Long.toString(Long.MAX_VALUE)); + producerProps.put(ProducerConfig.ACKS_CONFIG, "all"); + producerProps.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1"); producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.toString(Integer.MAX_VALUE)); producerProps.put(ProducerConfig.CLIENT_ID_CONFIG, defaultClientId); // User-specified overrides diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java index e57e5e33fea85..2b210794207aa 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java @@ -213,6 +213,8 @@ public void setup() { defaultProducerConfigs.put( ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer"); defaultProducerConfigs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, Long.toString(Long.MAX_VALUE)); + defaultProducerConfigs.put(ProducerConfig.ACKS_CONFIG, "all"); + defaultProducerConfigs.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1"); defaultProducerConfigs.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.toString(Integer.MAX_VALUE)); defaultConsumerConfigs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");