kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rha...@apache.org
Subject [kafka] branch 2.6 updated: KAFKA-9570: Define SSL configs in all worker config classes, not just distributed (#8135)
Date Fri, 05 Jun 2020 21:15:30 GMT
This is an automated email from the ASF dual-hosted git repository.

rhauch pushed a commit to branch 2.6
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/2.6 by this push:
     new 0567334  KAFKA-9570: Define SSL configs in all worker config classes, not just distributed
(#8135)
0567334 is described below

commit 0567334214966a342f5355840c982ac455c3b0b9
Author: Chris Egerton <chrise@confluent.io>
AuthorDate: Fri Jun 5 14:02:17 2020 -0700

    KAFKA-9570: Define SSL configs in all worker config classes, not just distributed (#8135)
    
    Define SSL configs in all worker config classes, not just distributed
    
    Author: Chris Egerton <chrise@confluent.io>
    Reviewers: Nigel Liang <nigel@nigelliang.com>, Randall Hauch <rhauch@gmail.com>
---
 .../apache/kafka/connect/runtime/WorkerConfig.java |  4 +-
 .../runtime/distributed/DistributedConfig.java     |  1 -
 .../runtime/standalone/StandaloneConfigTest.java   | 88 ++++++++++++++++++++++
 3 files changed, 91 insertions(+), 2 deletions(-)

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java
index 3217752..1b31a5f 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java
@@ -350,7 +350,9 @@ public class WorkerConfig extends AbstractConfig {
                 .define(TOPIC_CREATION_ENABLE_CONFIG, Type.BOOLEAN, TOPIC_CREATION_ENABLE_DEFAULT,
Importance.LOW,
                         TOPIC_CREATION_ENABLE_DOC)
                 .define(RESPONSE_HTTP_HEADERS_CONFIG, Type.STRING, RESPONSE_HTTP_HEADERS_DEFAULT,
-                        new ResponseHttpHeadersValidator(), Importance.LOW, RESPONSE_HTTP_HEADERS_DOC);
+                        new ResponseHttpHeadersValidator(), Importance.LOW, RESPONSE_HTTP_HEADERS_DOC)
+                // security support
+                .withClientSslSupport();
     }
 
     private void logInternalConverterDeprecationWarnings(Map<String, String> props)
{
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java
index 9af6291..4d67c06 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java
@@ -272,7 +272,6 @@ public class DistributedConfig extends WorkerConfig {
                     CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL,
                     ConfigDef.Importance.MEDIUM,
                     CommonClientConfigs.SECURITY_PROTOCOL_DOC)
-            .withClientSslSupport()
             .withClientSaslSupport()
             .define(WORKER_SYNC_TIMEOUT_MS_CONFIG,
                     ConfigDef.Type.INT,
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneConfigTest.java
b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneConfigTest.java
new file mode 100644
index 0000000..e2e886f
--- /dev/null
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneConfigTest.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.connect.runtime.standalone;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.types.Password;
+import org.apache.kafka.connect.runtime.WorkerConfig;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+
+public class StandaloneConfigTest {
+
+    private static final String HTTPS_LISTENER_PREFIX = "listeners.https.";
+
+    private Map<String, Object> sslProps() {
+        return new HashMap<String, Object>() {
+            {
+                put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, new Password("ssl_key_password"));
+                put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, "ssl_keystore");
+                put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, new Password("ssl_keystore_password"));
+                put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "ssl_truststore");
+                put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, new Password("ssl_truststore_password"));
+            }
+        };
+    }
+
+    private Map<String, String> baseWorkerProps() {
+        return new HashMap<String, String>() {
+            {
+                put(WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonConverter");
+                put(WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonConverter");
+                put(StandaloneConfig.OFFSET_STORAGE_FILE_FILENAME_CONFIG, "/tmp/foo");
+            }
+        };
+    }
+
+    private static Map<String, String> withStringValues(Map<String, ?> inputs,
String prefix) {
+        return ConfigDef.convertToStringMapWithPasswordValues(inputs).entrySet().stream()
+            .collect(Collectors.toMap(
+                entry -> prefix + entry.getKey(),
+                Map.Entry::getValue
+            ));
+    }
+
+    @Test
+    public void testRestServerPrefixedSslConfigs() {
+        Map<String, String> workerProps = baseWorkerProps();
+        Map<String, Object> expectedSslProps = sslProps();
+        workerProps.putAll(withStringValues(expectedSslProps, HTTPS_LISTENER_PREFIX));
+
+        StandaloneConfig config = new StandaloneConfig(workerProps);
+        assertEquals(expectedSslProps, config.valuesWithPrefixAllOrNothing(HTTPS_LISTENER_PREFIX));
+    }
+
+    @Test
+    public void testRestServerNonPrefixedSslConfigs() {
+        Map<String, String> props = baseWorkerProps();
+        Map<String, Object> expectedSslProps = sslProps();
+        props.putAll(withStringValues(expectedSslProps, ""));
+
+        StandaloneConfig config = new StandaloneConfig(props);
+        Map<String, Object> actualProps = config.valuesWithPrefixAllOrNothing(HTTPS_LISTENER_PREFIX)
+            .entrySet().stream()
+            .filter(entry -> expectedSslProps.containsKey(entry.getKey()))
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+        assertEquals(expectedSslProps, actualProps);
+    }
+}


Mime
View raw message