NPE when getting command topic configs

[stevenpyzhang]

Describe the bug
When we generate a map of topic configs in https://github.com/confluentinc/ksql/blob/master/ksqldb-engine/src/main/java/io/confluent/ksql/services/KafkaTopicClientImpl.java#L405

We don't do any null checks on the values of the configs, so if there are any, Collectors.toMap throws the NPE
https://bugs.openjdk.java.net/browse/JDK-8148463

We can get null values as the config value if the isSensitive flag is true https://kafka.apache.org/27/javadoc/org/apache/kafka/clients/admin/ConfigEntry.html#value--

To Reproduce
Set a sensitive topic config on the command topic

kafka-configs --bootstrap-server <KAFKA_SERVER> --entity-type topics --entity-name <command-topic-name> --describe --all
...
confluent.ssl.keystore.password=null sensitive=true synonyms={}
confluent.ssl.truststore.location=null sensitive=true synonyms={}
confluent.ssl.key.password=null sensitive=true synonyms={}
confluent.ssl.keystore.type=null sensitive=true synonyms={}
confluent.ssl.truststore.type=null sensitive=true synonyms={}
confluent.ssl.keystore.location=null sensitive=true synonyms={}
confluent.ssl.truststore.password=null sensitive=true synonyms={}
...

Try to start up server

Expected behavior
Server starts up normally

Actual behaviour
Server fails to start up with NPE

Additional context
A workaround might be to run ./bin/kafka-configs.sh --zookeeper <zookeeper> --alter \ --entity-type topics --entity-name <topic> --delete-config <config> and remove the sensitive configs

[stevenpyzhang]

Looks like there's a bug in the broker s.t. it returns broker configs as topic configs due to improper config filtering, so this isn't really a common issue. It still may be worth adding the null check in though as a safeguard

[skalinets]

So basically it means that ksqldb (at least 0.14.0) doesn't work with confluent brokers with enabled ssl? @stevenpyzhang could you please provide more details of workaround? How to remove sensitive configs from a topic (what should be substituted instead of <config>)? I tried passing eg. confluent.ssl.keystore.password -- it didn't work (complaining that this setting did not exist).

[stevenpyzhang]

@skalinets I'm not sure there is a workaround currently as those configs are actually broker level configs that are being mistakenly returned as TopicConfigs from the server. I think the only workaround would be to use a KSQL version that has the null pointer check. I'll try to fix this as soon as possible and look into backporting this to 0.14.0

[DmytroMelnyk]

Could you, please, merge it into 0.15.0?