[#2941] Fix handling of empty string Kafka config property values.

Signed-off-by: Carsten Lohmann <[email protected]>

adapter-base-quarkus/src/test/java/org/eclipse/hono/adapter/quarkus/QuarkusPropertyBindingTest.java

@@ -51,6 +51,7 @@ public void testThatClientOptionsExist() {
     @Test
     public void testThatCommonConfigIsPresent() {
         assertThat(kafkaClientOptions.commonClientConfig().get("common.property")).isEqualTo("present");
+        assertThat(kafkaClientOptions.commonClientConfig().get("empty")).isEqualTo("");
     }
 
     /**

adapter-base-quarkus/src/test/resources/application.yaml

@@ -7,6 +7,7 @@ hono:
     commonClientConfig:
       common.property: "present"
       number: 123
+      empty: ""
       bootstrap.servers: "example.com:9999"
       admin.property: "common"
       consumer.property: "common"

clients/kafka-common/pom.xml

@@ -70,6 +70,11 @@
       <artifactId>smallrye-config</artifactId>
       <optional>true</optional>
     </dependency>
+    <dependency>
+      <groupId>jakarta.enterprise</groupId>
+      <artifactId>jakarta.enterprise.cdi-api</artifactId>
+      <optional>true</optional>
+    </dependency>
 
     <!-- Testing -->
     <dependency>

clients/kafka-common/src/main/java/org/eclipse/hono/client/kafka/KafkaClientOptions.java

@@ -11,44 +11,100 @@
  * SPDX-License-Identifier: EPL-2.0
  */
 
-
 package org.eclipse.hono.client.kafka;
 
 import java.util.Map;
+import java.util.stream.Collectors;
+
+import javax.enterprise.context.ApplicationScoped;
+import javax.inject.Inject;
 
 import io.smallrye.config.ConfigMapping;
+import io.smallrye.config.ConfigValue;
 
 /**
  * Common options for configuring Kafka clients.
  */
-@ConfigMapping(prefix = "hono.kafka", namingStrategy = ConfigMapping.NamingStrategy.VERBATIM)
-public interface KafkaClientOptions {
+@ApplicationScoped
+public class KafkaClientOptions {
+
+    @Inject
+    RawKafkaClientOptions rawKafkaClientOptions;
 
     /**
      * The properties shared by all types of clients.
      *
      * @return The properties.
      */
-    Map<String, String> commonClientConfig();
+    public Map<String, String> commonClientConfig() {
+        return toStringValueMap(rawKafkaClientOptions.commonClientConfig());
+    }
 
     /**
      * The properties to use for admin clients.
      *
      * @return The properties.
      */
-    Map<String, String> adminClientConfig();
+    public Map<String, String> adminClientConfig() {
+        return toStringValueMap(rawKafkaClientOptions.adminClientConfig());
+    }
 
     /**
      * The properties to use for consumers.
      *
      * @return The properties.
      */
-    Map<String, String> consumerConfig();
+    public Map<String, String> consumerConfig() {
+        return toStringValueMap(rawKafkaClientOptions.consumerConfig());
+    }
 
     /**
      * The properties to use for producers.
      *
      * @return The properties.
      */
-    Map<String, String> producerConfig();
+    public Map<String, String> producerConfig() {
+        return toStringValueMap(rawKafkaClientOptions.producerConfig());
+    }
+
+    private static Map<String, String> toStringValueMap(final Map<String, ConfigValue> config) {
+        return config.entrySet().stream()
+                .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().getValue()));
+    }
+
+    /**
+     * Container for the Kafka client configuration Maps.
+     * <p>
+     * This class with its {@link ConfigValue} (instead of String) Map values is needed to support empty String values.
+     * The default value converters consider an empty String as a {@code null} value and then throw a
+     * {@link java.util.NoSuchElementException}.
+     * Using {@link ConfigValue} bypasses usage of the converter and allows access to the original value.
+     */
+    @ConfigMapping(prefix = "hono.kafka", namingStrategy = ConfigMapping.NamingStrategy.VERBATIM)
+    interface RawKafkaClientOptions {
+
+        /**
+         * The properties shared by all types of clients.
+         * @return The properties.
+         */
+        Map<String, ConfigValue> commonClientConfig();
+
+        /**
+         * The properties to use for admin clients.
+         * @return The properties.
+         */
+        Map<String, ConfigValue> adminClientConfig();
+
+        /**
+         * The properties to use for consumers.
+         * @return The properties.
+         */
+        Map<String, ConfigValue> consumerConfig();
+
+        /**
+         * The properties to use for producers.
+         * @return The properties.
+         */
+        Map<String, ConfigValue> producerConfig();
+    }
 }