fix: filter null entries before creating KafkaConfigStore (#3147)

confluentinc · Jul 30, 2019 · 2852af1 · 2852af1
1 parent 377e961
commit 2852af1
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 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
@@ -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/...est-app/src/test/java/io/confluent/ksql/rest/server/computation/KafkaConfigStoreTest.java b/...est-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);