Add Service Binding configuration for Kafka

Fixes: quarkusio#15631

docs/src/main/asciidoc/deploying-to-kubernetes.adoc

@@ -1198,6 +1198,7 @@ Currently, the following Quarkus extensions support this feature:
 * quarkus-jdbc-mssql
 * quarkus-jdbc-mysql
 * quarkus-jdbc-postgresql
+* quarkus-kafka-client
 
 This list of extensions will grow as more services with supported bindings become available on Kubernetes.
 

extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java

@@ -64,6 +64,7 @@
 import io.quarkus.deployment.builditem.nativeimage.ReflectiveHierarchyBuildItem;
 import io.quarkus.deployment.builditem.nativeimage.ServiceProviderBuildItem;
 import io.quarkus.deployment.pkg.NativeConfig;
+import io.quarkus.kafka.client.runtime.KafkaBindingConverter;
 import io.quarkus.kafka.client.runtime.KafkaRecorder;
 import io.quarkus.kafka.client.runtime.KafkaRuntimeConfigProducer;
 import io.quarkus.kafka.client.serialization.JsonbDeserializer;
@@ -394,6 +395,16 @@ UnremovableBeanBuildItem ensureJsonParserAvailable() {
                 "javax.json.bind.Jsonb");
     }
 
+    @BuildStep
+    void registerServiceBinding(Capabilities capabilities,
+            BuildProducer<ServiceProviderBuildItem> serviceProvider) {
+        if (capabilities.isPresent(Capability.KUBERNETES_SERVICE_BINDING)) {
+            serviceProvider.produce(
+                    new ServiceProviderBuildItem("io.quarkus.kubernetes.service.binding.runtime.ServiceBindingConverter",
+                            KafkaBindingConverter.class.getName()));
+        }
+    }
+
     public static String getArch() {
         String osArch = System.getProperty("os.arch");
         return osArch.replaceAll("\\W", "");

extensions/kafka-client/runtime/pom.xml

@@ -33,6 +33,11 @@
             <artifactId>quarkus-smallrye-health</artifactId>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-kubernetes-service-binding</artifactId>
+            <optional>true</optional>
+        </dependency>
 
         <dependency>
             <groupId>io.quarkus</groupId>

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

@@ -0,0 +1,34 @@
+package io.quarkus.kafka.client.runtime;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import io.quarkus.kubernetes.service.binding.runtime.ServiceBinding;
+import io.quarkus.kubernetes.service.binding.runtime.ServiceBindingConfigSource;
+import io.quarkus.kubernetes.service.binding.runtime.ServiceBindingConverter;
+
+public class KafkaBindingConverter implements ServiceBindingConverter {
+
+    @Override
+    public Optional<ServiceBindingConfigSource> convert(List<ServiceBinding> serviceBindings) {
+        Optional<ServiceBinding> matchingByType = ServiceBinding.singleMatchingByType("kafka", serviceBindings);
+        if (!matchingByType.isPresent()) {
+            return Optional.empty();
+        }
+
+        Map<String, String> properties = new HashMap<>();
+        ServiceBinding binding = matchingByType.get();
+
+        String bootstrapServers = binding.getProperties().get("bootstrapServers");
+        if (bootstrapServers == null) {
+            bootstrapServers = binding.getProperties().get("bootstrap-servers");
+        }
+        if (bootstrapServers != null) {
+            properties.put("kafka.bootstrap.servers", bootstrapServers);
+        }
+
+        return Optional.of(new ServiceBindingConfigSource("kafka-k8s-service-binding-source", properties));
+    }
+}

extensions/kafka-client/runtime/src/main/resources/META-INF/services/io.quarkus.kubernetes.service.binding.runtime.ServiceBindingConverter

@@ -0,0 +1 @@
+io.quarkus.kafka.client.runtime.KafkaBindingConverter

integration-tests/kafka-snappy/pom.xml

@@ -49,6 +49,12 @@
             <artifactId>quarkus-kafka-client</artifactId>
         </dependency>
 
+        <!-- Service Binding - This isn't necessary to any Kafka functionality, but we want to test that Quarkus provides the proper config from a mock service binding -->
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-kubernetes-service-binding</artifactId>
+        </dependency>
+
         <!-- Micrometer -->
         <dependency>
             <groupId>io.quarkus</groupId>
@@ -116,6 +122,19 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-kubernetes-service-binding-deployment</artifactId>
+            <version>${project.version}</version>
+            <type>pom</type>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>*</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-resteasy-deployment</artifactId>

integration-tests/kafka-snappy/src/main/resources/application.properties

@@ -6,5 +6,3 @@ quarkus.kafka.health.enabled=true
 
 # Enable snappy:
 quarkus.kafka.snappy.enabled=true
-
-kafka.bootstrap.servers=localhost:19092

integration-tests/kafka-snappy/src/test/java/io/quarkus/it/kafka/KafkaTestResource.java

@@ -1,7 +1,8 @@
 package io.quarkus.it.kafka;
 
 import java.io.File;
-import java.util.Collections;
+import java.nio.file.Paths;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
 
@@ -29,7 +30,11 @@ public Map<String, String> start() {
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
-        return Collections.emptyMap();
+        Map<String, String> result = new HashMap<>();
+        // make the service binding root known
+        result.put("quarkus.kubernetes-service-binding.root", Paths.get("").resolve("src").resolve("test").resolve("resources")
+                .resolve("k8s-sb").toAbsolutePath().toString());
+        return result;
     }
 
     @Override

integration-tests/kafka-snappy/src/test/resources/k8s-sb/some-kafka/bootstrapServers

@@ -0,0 +1 @@
+localhost:19092

integration-tests/kafka-snappy/src/test/resources/k8s-sb/some-kafka/provider

@@ -0,0 +1 @@
+RH-SBO

integration-tests/kafka-snappy/src/test/resources/k8s-sb/some-kafka/type

@@ -0,0 +1 @@
+kafka