From 403d89edeaa7808f71c0e7318411c925895210f2 Mon Sep 17 00:00:00 2001
From: Ewen Cheslack-Postava <me@ewencp.org>
Date: Tue, 10 Nov 2015 11:07:26 -0800
Subject: [PATCH] KAFKA-2798: Use prefixedd configurations for Kafka Connect
 producer and consumer settings so they do not conflict with the distributed
 herder's settings.

Author: Ewen Cheslack-Postava <me@ewencp.org>

Reviewers: Gwen Shapira

Closes #486 from ewencp/kafka-2798-conflicting-herder-producer-consumer-configs
---
 .../org/apache/kafka/common/config/AbstractConfig.java    | 8 --------
 .../java/org/apache/kafka/connect/runtime/Worker.java     | 2 +-
 .../org/apache/kafka/connect/runtime/WorkerSinkTask.java  | 5 ++++-
 3 files changed, 5 insertions(+), 10 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java b/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
index 07b64c044d745..10293561ac851 100644
--- a/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
@@ -105,14 +105,6 @@ public Set<String> unused() {
         return keys;
     }
 
-    public Map<String, Object> unusedConfigs() {
-        Set<String> unusedKeys = this.unused();
-        Map<String, Object> unusedProps = new HashMap<>();
-        for (String key : unusedKeys)
-            unusedProps.put(key, this.originals.get(key));
-        return unusedProps;
-    }
-
     public Map<String, Object> originals() {
         Map<String, Object> copy = new RecordingMap<>();
         copy.putAll(originals);
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 359a79c8e1e8b..f5b23ecda6b7e 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
@@ -101,7 +101,7 @@ public void start() {
         producerProps.put(ProducerConfig.ACKS_CONFIG, "all");
         producerProps.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1");
 
-        producerProps.putAll(config.unusedConfigs());
+        producerProps.putAll(config.originalsWithPrefix("producer."));
 
         producer = new KafkaProducer<>(producerProps);
 
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
index 643b10e25acfb..e0a3e04b639f0 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
@@ -233,7 +233,8 @@ public WorkerConfig workerConfig() {
     private KafkaConsumer<byte[], byte[]> createConsumer() {
         // Include any unknown worker configs so consumer configs can be set globally on the worker
         // and through to the task
-        Map<String, Object> props = workerConfig.unusedConfigs();
+        Map<String, Object> props = new HashMap<>();
+
         props.put(ConsumerConfig.GROUP_ID_CONFIG, "connect-" + id.connector());
         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                 Utils.join(workerConfig.getList(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG), ","));
@@ -242,6 +243,8 @@ private KafkaConsumer<byte[], byte[]> createConsumer() {
         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
 
+        props.putAll(workerConfig.originalsWithPrefix("consumer."));
+
         KafkaConsumer<byte[], byte[]> newConsumer;
         try {
             newConsumer = new KafkaConsumer<>(props);