From 2852af1059a3181b20fc016429c17acb3f1dfb71 Mon Sep 17 00:00:00 2001
From: Almog Gavra <almog@confluent.io>
Date: Tue, 30 Jul 2019 11:14:36 -0700
Subject: [PATCH] fix: filter null entries before creating KafkaConfigStore
 (#3147)

---
 .../ksql/rest/server/computation/KafkaConfigStore.java     | 7 ++++++-
 .../ksql/rest/server/computation/KafkaConfigStoreTest.java | 7 +++----
 2 files changed, 9 insertions(+), 5 deletions(-)

diff --git a/ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/KafkaConfigStore.java b/ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/KafkaConfigStore.java
index 80b6b0f2ff8c..cee851e2d6f6 100644
--- a/ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/KafkaConfigStore.java
+++ b/ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/KafkaConfigStore.java
@@ -26,6 +26,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.function.Supplier;
@@ -116,7 +117,11 @@ public static class KsqlProperties {
     KsqlProperties(
         @JsonProperty("ksqlProperties") final Map<String, String> ksqlProperties) {
       this.ksqlProperties = ksqlProperties == null
-          ? Collections.emptyMap() : ImmutableMap.copyOf(ksqlProperties);
+          ? Collections.emptyMap()
+          : ksqlProperties.entrySet()
+              .stream()
+              .filter(kv -> kv.getValue() != null)
+              .collect(ImmutableMap.toImmutableMap(Entry::getKey, Entry::getValue));
     }
 
     public Map<String, String> getKsqlProperties() {
diff --git a/ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/computation/KafkaConfigStoreTest.java b/ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/computation/KafkaConfigStoreTest.java
index 4d4202c7871a..ca1286e9ab7c 100644
--- a/ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/computation/KafkaConfigStoreTest.java
+++ b/ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/computation/KafkaConfigStoreTest.java
@@ -72,13 +72,13 @@ public class KafkaConfigStoreTest {
       ImmutableMap.of(KsqlConfig.KSQL_PERSISTENT_QUERY_NAME_PREFIX_CONFIG, "bad"));
 
   private final KsqlProperties properties = new KsqlProperties(
-      filterNullValues(currentConfig.getAllConfigPropsWithSecretsObfuscated())
+      currentConfig.getAllConfigPropsWithSecretsObfuscated()
   );
   private final KsqlProperties savedProperties = new KsqlProperties(
-      filterNullValues(savedConfig.getAllConfigPropsWithSecretsObfuscated())
+      savedConfig.getAllConfigPropsWithSecretsObfuscated()
   );
   private final KsqlProperties badProperties = new KsqlProperties(
-      filterNullValues(badConfig.getAllConfigPropsWithSecretsObfuscated())
+      badConfig.getAllConfigPropsWithSecretsObfuscated()
   );
 
   private final TopicPartition topicPartition = new TopicPartition(TOPIC_NAME, 0);
@@ -107,7 +107,6 @@ public class KafkaConfigStoreTest {
   private InOrder inOrder;
 
   @Before
-  @SuppressWarnings("unchecked")
   public void setUp() {
     when(producerSupplier.get()).thenReturn(producer);
     when(consumerSupplier.get()).thenReturn(consumerBefore).thenReturn(consumerAfter);