fix: npe when getting topic configs (#6946)

confluentinc · Feb 4, 2021 · 5e026d4 · 5e026d4
1 parent 96288e4
commit 5e026d4
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 0 deletions.
diff --git a/ksqldb-engine/src/main/java/io/confluent/ksql/services/KafkaTopicClientImpl.java b/ksqldb-engine/src/main/java/io/confluent/ksql/services/KafkaTopicClientImpl.java
@@ -401,6 +401,7 @@ private Map<String, String> topicConfig(
           () -> adminClient.get().describeConfigs(request).all().get(),
           ExecutorUtil.RetryBehaviour.ON_RETRYABLE).get(resource);
       return config.entries().stream()
+          .filter(e -> e.value() != null)
           .filter(e -> includeDefaults || !e.isDefault())
           .collect(Collectors.toMap(ConfigEntry::name, ConfigEntry::value));
     } catch (final Exception e) {

diff --git a/ksqldb-engine/src/test/java/io/confluent/ksql/services/KafkaTopicClientImplTest.java b/ksqldb-engine/src/test/java/io/confluent/ksql/services/KafkaTopicClientImplTest.java
@@ -534,6 +534,23 @@ public void shouldGetTopicConfig() {
     assertThat(config.get(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG), is("1"));
   }
 
+  @Test
+  public void shouldNotFailWhenTopicConfigValueIsNull() {
+    // Given:
+    givenTopicConfigs(
+        "fred",
+        overriddenConfigEntry(TopicConfig.RETENTION_MS_CONFIG, "12345"),
+        overriddenConfigEntry(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, null)
+    );
+
+    // When:
+    final Map<String, String> config = kafkaTopicClient.getTopicConfig("fred");
+
+    // Then:
+    assertThat(config.get(TopicConfig.RETENTION_MS_CONFIG), is("12345"));
+    assertThat(config.containsKey(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG), is(false));
+  }
+
   @Test
   public void shouldGetTopicCleanUpPolicyDelete() {
     // Given: