Filter out Properties passed into StreamsRegistry

Author: wavejumper

src/java/io/operatr/kpow/StreamsRegistry.java

@@ -7,6 +7,7 @@
 import org.apache.kafka.streams.KafkaStreams;
 import org.apache.kafka.streams.Topology;
 
+import java.util.ArrayList;
 import java.util.Properties;
 
 public class StreamsRegistry implements AutoCloseable {
@@ -25,6 +26,40 @@ public String getId() {
 
     private final Object agent;
 
+    public static Properties filterProperties(Properties props) {
+        ArrayList<String> allowedKeys = new ArrayList<>();
+        allowedKeys.add("security.protocol");
+        allowedKeys.add("sasl.mechanism");
+        allowedKeys.add("sasl.jaas.config");
+        allowedKeys.add("sasl.login.callback.handler.class");
+        allowedKeys.add("ssl.keystore.location");
+        allowedKeys.add("ssl.keystore.password");
+        allowedKeys.add("ssl.key.password");
+        allowedKeys.add("ssl.keystore.type");
+        allowedKeys.add("ssl.keymanager.algorithm");
+        allowedKeys.add("ssl.truststore.location");
+        allowedKeys.add("ssl.truststore.password");
+        allowedKeys.add("ssl.truststore.type");
+        allowedKeys.add("ssl.trustmanager.algorithm");
+        allowedKeys.add("ssl.endpoint.identification.algorithm");
+        allowedKeys.add("ssl.provider");
+        allowedKeys.add("ssl.cipher.suites");
+        allowedKeys.add("ssl.protocol");
+        allowedKeys.add("ssl.enabled.protocols");
+        allowedKeys.add("ssl.secure.random.implementation");
+        allowedKeys.add("ssl.keystore.key");
+        allowedKeys.add("ssl.keystore.certificate.chain");
+        allowedKeys.add("ssl.truststore.certificates");
+        allowedKeys.add("bootstrap.servers");
+        Properties nextProps = new Properties();
+        for (String key : allowedKeys) {
+            if (props.containsKey(key)) {
+                nextProps.setProperty(key, String.valueOf(props.get(key)));
+            }
+        }
+        return nextProps;
+    }
+
     public StreamsRegistry(Properties props) {
         IFn require = Clojure.var("clojure.core", "require");
         require.invoke(Clojure.read("io.operatr.kpow.agent"));
@@ -33,7 +68,8 @@ public StreamsRegistry(Properties props) {
         IFn serdesFn = Clojure.var("io.operatr.kpow.serdes", "transit-json-serializer");
         Serializer keySerializer = (Serializer) serdesFn.invoke();
         Serializer valSerializer = (Serializer) serdesFn.invoke();
-        KafkaProducer producer = new KafkaProducer<>(props, keySerializer, valSerializer);
+        Properties producerProps = filterProperties(props);
+        KafkaProducer producer = new KafkaProducer<>(producerProps, keySerializer, valSerializer);
         agent = agentFn.invoke(producer);
     }
 

test/io/operatr/agent_test.clj

@@ -4,7 +4,20 @@
             [io.operatr.kpow.agent :as agent])
   (:import (org.apache.kafka.streams Topology StreamsBuilder KafkaStreams$State)
            (org.apache.kafka.common MetricName Metric)
-           (org.apache.kafka.clients.producer Producer)))
+           (org.apache.kafka.clients.producer Producer)
+           (io.operatr.kpow StreamsRegistry)
+           (java.util Properties)))
+
+(defn ^Properties ->props [m]
+  (let [props (Properties.)]
+    (doseq [[k v] m]
+      (.put props k v))
+    props))
+
+(deftest filter-props
+  (is (empty? (StreamsRegistry/filterProperties (->props {"fo" "bar"}))))
+  (is (= {"bootstrap.servers" "xyz"}
+         (into {} (StreamsRegistry/filterProperties (->props {"fo" "bar" "bootstrap.servers" "xyz"}))))))
 
 (defn ^Topology test-topology
   []