You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by kk...@apache.org on 2022/03/22 22:19:53 UTC
[kafka] branch revert-11475-KAFKA-7077_KIP-318 created (now 8167da9)
This is an automated email from the ASF dual-hosted git repository.
kkarantasis pushed a change to branch revert-11475-KAFKA-7077_KIP-318
in repository https://gitbox.apache.org/repos/asf/kafka.git.
at 8167da9 Revert "KAFKA-7077: Use default producer settings in Connect Worker (#11475)"
This branch includes the following new commits:
new 8167da9 Revert "KAFKA-7077: Use default producer settings in Connect Worker (#11475)"
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[kafka] 01/01: Revert "KAFKA-7077: Use default producer settings in Connect Worker (#11475)"
Posted by kk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kkarantasis pushed a commit to branch revert-11475-KAFKA-7077_KIP-318
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 8167da97bf2fc15b6dddee1465f41030cd128ca1
Author: Konstantine Karantasis <ko...@confluent.io>
AuthorDate: Tue Mar 22 15:17:51 2022 -0700
Revert "KAFKA-7077: Use default producer settings in Connect Worker (#11475)"
This reverts commit 76cf7a5793702b55e2cfd98a375f8f1708ff32c3.
---
.../runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java | 2 ++
.../src/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 45c7d9f..582271a 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 @@ public class Worker {
// 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 e57e5e3..2b21079 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 class WorkerTest extends ThreadedTest {
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");