Fix KAFKA_TLS_CONFIGURATION_NAME environment variable not being recognized

Fixes quarkusio#51195

Author: ozangunalp

extensions/kafka-client/runtime/pom.xml

@@ -78,6 +78,11 @@
             <artifactId>assertj-core</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

extensions/kafka-client/runtime/src/main/java/io/quarkus/kafka/client/runtime/KafkaRuntimeConfigProducer.java

@@ -17,6 +17,7 @@
 public class KafkaRuntimeConfigProducer {
 
     public static final String TLS_CONFIG_NAME_KEY = "tls-configuration-name";
+    private static final String TLS_CONFIG_NAME_KEY_NORMALIZED = "tls.configuration.name";
 
     // not "kafka.", because we also inspect env vars, which start with "KAFKA_"
     private static final String CONFIG_PREFIX = "kafka";
@@ -46,9 +47,15 @@ public Map<String, Object> createKafkaRuntimeConfig(Config config, ApplicationCo
             String effectivePropertyName = propertyNameLowerCase.substring(CONFIG_PREFIX.length() + 1).toLowerCase()
                     .replace("_", ".");
             String value = config.getOptionalValue(propertyName, String.class).orElse("");
-            result.put(effectivePropertyName, value);
-            if (effectivePropertyName.equals(TLS_CONFIG_NAME_KEY)) {
+            // Normalize both property names for comparison (replace hyphens with dots)
+            // This ensures both YAML properties (kafka.tls-configuration-name) and
+            // environment variables (KAFKA_TLS_CONFIGURATION_NAME) are matched correctly
+            if (effectivePropertyName.equals(TLS_CONFIG_NAME_KEY) ||
+                    effectivePropertyName.equals(TLS_CONFIG_NAME_KEY_NORMALIZED)) {
+                result.put(TLS_CONFIG_NAME_KEY, value);
                 result.put("ssl.engine.factory.class", QuarkusKafkaSslEngineFactory.class.getName());
+            } else {
+                result.put(effectivePropertyName, value);
             }
         }
 

extensions/kafka-client/runtime/src/test/java/io/quarkus/kafka/client/runtime/KafkaRuntimeConfigProducerTest.java

@@ -0,0 +1,87 @@
+package io.quarkus.kafka.client.runtime;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Map;
+import java.util.Optional;
+
+import org.eclipse.microprofile.config.Config;
+import org.junit.jupiter.api.Test;
+
+import io.quarkus.kafka.client.tls.QuarkusKafkaSslEngineFactory;
+import io.quarkus.runtime.ApplicationConfig;
+import io.smallrye.config.SmallRyeConfig;
+import io.smallrye.config.SmallRyeConfigBuilder;
+
+public class KafkaRuntimeConfigProducerTest {
+
+    @Test
+    public void testTlsConfigurationNameWithHyphens() {
+        // Test when property is set via application.yaml as kafka.tls-configuration-name
+        Config config = new SmallRyeConfigBuilder()
+                .withDefaultValues(Map.of("kafka.tls-configuration-name", "my-tls-config"))
+                .build();
+        ApplicationConfig appConfig = mock(ApplicationConfig.class);
+        when(appConfig.name()).thenReturn(Optional.empty());
+
+        KafkaRuntimeConfigProducer producer = new KafkaRuntimeConfigProducer();
+        Map<String, Object> result = producer.createKafkaRuntimeConfig(config, appConfig);
+
+        assertThat(result)
+                .containsEntry("tls-configuration-name", "my-tls-config")
+                .containsEntry("ssl.engine.factory.class", QuarkusKafkaSslEngineFactory.class.getName());
+    }
+
+    @Test
+    public void testTlsConfigurationNameWithUnderscores() {
+        // Test when property is set via environment variable as KAFKA_TLS_CONFIGURATION_NAME
+        Config config = new SmallRyeConfigBuilder()
+                .withDefaultValues(Map.of("KAFKA_TLS_CONFIGURATION_NAME", "my-tls-config"))
+                .build();
+        ApplicationConfig appConfig = mock(ApplicationConfig.class);
+        when(appConfig.name()).thenReturn(Optional.empty());
+
+        KafkaRuntimeConfigProducer producer = new KafkaRuntimeConfigProducer();
+        Map<String, Object> result = producer.createKafkaRuntimeConfig(config, appConfig);
+
+        assertThat(result)
+                .containsEntry("tls-configuration-name", "my-tls-config")
+                .containsEntry("ssl.engine.factory.class", QuarkusKafkaSslEngineFactory.class.getName());
+    }
+
+    @Test
+    public void testOtherKafkaProperties() {
+        // Test that other Kafka properties are processed correctly
+        Config config = new SmallRyeConfigBuilder()
+                .withDefaultValues(Map.of("kafka.bootstrap.servers", "localhost:9092",
+                        "KAFKA_CONSUMER_MAX_POLL_RECORDS", "100"))
+                .build();
+        ApplicationConfig appConfig = mock(ApplicationConfig.class);
+        when(appConfig.name()).thenReturn(Optional.empty());
+
+        KafkaRuntimeConfigProducer producer = new KafkaRuntimeConfigProducer();
+        Map<String, Object> result = producer.createKafkaRuntimeConfig(config, appConfig);
+
+        assertThat(result)
+                .containsEntry("bootstrap.servers", "localhost:9092")
+                .containsEntry("consumer.max.poll.records", "100");
+    }
+
+    @Test
+    public void testGroupIdDefaultsToApplicationName() {
+        SmallRyeConfig config = new SmallRyeConfigBuilder()
+                .withDefaultValues(Map.of("kafka.bootstrap.servers", "localhost:9092"))
+                .build();
+        ApplicationConfig appConfig = mock(ApplicationConfig.class);
+        when(appConfig.name()).thenReturn(Optional.of("my-app"));
+
+        KafkaRuntimeConfigProducer producer = new KafkaRuntimeConfigProducer();
+        Map<String, Object> result = producer.createKafkaRuntimeConfig(config, appConfig);
+
+        assertThat(result)
+                .containsEntry("bootstrap.servers", "localhost:9092")
+                .containsEntry("group.id", "my-app");
+    }
+}