Add Quarkus service binding support to Kafka extension

Fixes apache#2333

extensions/kafka/deployment/src/main/java/org/apache/camel/quarkus/component/kafka/deployment/KafkaProcessor.java

@@ -16,8 +16,16 @@
  */
 package org.apache.camel.quarkus.component.kafka.deployment;
 
+import java.util.List;
+
 import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.annotations.ExecutionTime;
+import io.quarkus.deployment.annotations.Record;
 import io.quarkus.deployment.builditem.FeatureBuildItem;
+import io.quarkus.deployment.builditem.nativeimage.ServiceProviderBuildItem;
+import org.apache.camel.component.kafka.KafkaComponent;
+import org.apache.camel.quarkus.component.kafka.CamelKafkaRecorder;
+import org.apache.camel.quarkus.core.deployment.spi.CamelRuntimeBeanBuildItem;
 
 class KafkaProcessor {
     private static final String FEATURE = "camel-kafka";
@@ -26,4 +34,16 @@ class KafkaProcessor {
     FeatureBuildItem feature() {
         return new FeatureBuildItem(FEATURE);
     }
+
+    @BuildStep
+    @Record(ExecutionTime.RUNTIME_INIT)
+    CamelRuntimeBeanBuildItem createCamelKafkaComponent(
+            CamelKafkaRecorder recorder,
+            // We want Quarkus to configure the ServiceBindingConverter bits before this step
+            List<ServiceProviderBuildItem> serviceProviders) {
+        return new CamelRuntimeBeanBuildItem(
+                "kafka",
+                KafkaComponent.class.getName(),
+                recorder.createKafkaComponent());
+    }
 }

extensions/kafka/runtime/src/main/java/org/apache/camel/quarkus/component/kafka/CamelKafkaRecorder.java

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.kafka;
+
+import java.util.Collections;
+import java.util.Map;
+
+import io.quarkus.arc.Arc;
+import io.quarkus.arc.InstanceHandle;
+import io.quarkus.runtime.RuntimeValue;
+import io.quarkus.runtime.annotations.Recorder;
+import org.apache.camel.component.kafka.KafkaComponent;
+
+@Recorder
+public class CamelKafkaRecorder {
+
+    @SuppressWarnings("unchecked")
+    public RuntimeValue<KafkaComponent> createKafkaComponent() {
+        final KafkaComponent component = new KafkaComponent();
+        final InstanceHandle<Object> instance = Arc.container().instance("default-kafka-broker");
+        Map<String, Object> kafkaConfig;
+
+        if (instance.isAvailable()) {
+            kafkaConfig = (Map<String, Object>) instance.get();
+        } else {
+            kafkaConfig = Collections.emptyMap();
+        }
+
+        // TODO: Return new RuntimeValue<>(quarkusKafkaClientFactory) as the KafkaClientFactory option should be autowired
+        // https://issues.apache.org/jira/browse/CAMEL-16500
+        QuarkusKafkaClientFactory quarkusKafkaClientFactory = new QuarkusKafkaClientFactory(kafkaConfig);
+        component.setKafkaClientFactory(quarkusKafkaClientFactory);
+        return new RuntimeValue<>(component);
+    }
+}

extensions/kafka/runtime/src/main/java/org/apache/camel/quarkus/component/kafka/QuarkusKafkaClientFactory.java

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.kafka;
+
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.camel.component.kafka.DefaultKafkaClientFactory;
+import org.apache.camel.component.kafka.KafkaConfiguration;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+
+/**
+ * Custom {@link org.apache.camel.component.kafka.KafkaClientFactory} to enable Kafka configuration properties
+ * discovered by Quarkus to be merged with those configured from the Camel Kafka component and endpoint URI options.
+ */
+public class QuarkusKafkaClientFactory extends DefaultKafkaClientFactory {
+
+    private final Map<String, Object> quarkusKafkaConfiguration;
+
+    public QuarkusKafkaClientFactory(Map<String, Object> quarkusKafkaConfiguration) {
+        this.quarkusKafkaConfiguration = quarkusKafkaConfiguration;
+    }
+
+    @Override
+    public KafkaProducer getProducer(Properties camelKafkaProperties) {
+        mergeConfiguration(camelKafkaProperties);
+        return super.getProducer(camelKafkaProperties);
+    }
+
+    @Override
+    public KafkaConsumer getConsumer(Properties camelKafkaProperties) {
+        mergeConfiguration(camelKafkaProperties);
+        return super.getConsumer(camelKafkaProperties);
+    }
+
+    @Override
+    public String getBrokers(KafkaConfiguration configuration) {
+        String brokers = (String) quarkusKafkaConfiguration.get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG);
+        return brokers != null ? brokers : super.getBrokers(configuration);
+    }
+
+    /**
+     * Merges kafka configuration properties discovered by Quarkus with those provided via the
+     * component & endpoint URI options.
+     */
+    private void mergeConfiguration(Properties camelKafkaProperties) {
+        if (quarkusKafkaConfiguration != null) {
+            for (Map.Entry<String, Object> entry : quarkusKafkaConfiguration.entrySet()) {
+                camelKafkaProperties.put(entry.getKey(), entry.getValue());
+            }
+        }
+    }
+}

integration-tests/kafka-sasl/pom.xml

@@ -0,0 +1,134 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one or more
+    contributor license agreements.  See the NOTICE file distributed with
+    this work for additional information regarding copyright ownership.
+    The ASF licenses this file to You under the Apache License, Version 2.0
+    (the "License"); you may not use this file except in compliance with
+    the License.  You may obtain a copy of the License at
+
+         http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.camel.quarkus</groupId>
+        <artifactId>camel-quarkus-integration-tests</artifactId>
+        <version>1.9.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>camel-quarkus-integration-test-kafka-sasl</artifactId>
+    <name>Camel Quarkus :: Integration Tests :: Kafka SASL</name>
+    <description>Integration tests for Camel Quarkus Kafka SASL</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.camel.quarkus</groupId>
+            <artifactId>camel-quarkus-kafka</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.camel.quarkus</groupId>
+            <artifactId>camel-quarkus-log</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-resteasy</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-resteasy-jsonb</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-jackson</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-kubernetes-service-binding</artifactId>
+        </dependency>
+
+        <!-- test dependencies -->
+        <dependency>
+            <groupId>io.rest-assured</groupId>
+            <artifactId>rest-assured</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.camel.quarkus</groupId>
+            <artifactId>camel-quarkus-integration-testcontainers-support</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>kafka</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <!-- The following dependencies guarantee that this module is built after them. You can update them by running `mvn process-resources -Pformat -N` from the source tree root directory -->
+        <dependency>
+            <groupId>org.apache.camel.quarkus</groupId>
+            <artifactId>camel-quarkus-kafka-deployment</artifactId>
+            <version>${project.version}</version>
+            <type>pom</type>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>*</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.camel.quarkus</groupId>
+            <artifactId>camel-quarkus-log-deployment</artifactId>
+            <version>${project.version}</version>
+            <type>pom</type>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>*</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+    </dependencies>
+
+    <profiles>
+        <profile>
+            <id>native</id>
+            <activation>
+                <property>
+                    <name>native</name>
+                </property>
+            </activation>
+            <properties>
+                <quarkus.package.type>native</quarkus.package.type>
+            </properties>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.apache.maven.plugins</groupId>
+                        <artifactId>maven-failsafe-plugin</artifactId>
+                        <executions>
+                            <execution>
+                                <goals>
+                                    <goal>integration-test</goal>
+                                    <goal>verify</goal>
+                                </goals>
+                            </execution>
+                        </executions>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+    </profiles>
+
+</project>

integration-tests/kafka-sasl/src/main/java/org/apache/camel/quarkus/kafka/sasl/KafkaSaslResource.java

@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.kafka.sasl;
+
+import java.time.Duration;
+
+import javax.enterprise.context.ApplicationScoped;
+import javax.json.Json;
+import javax.json.JsonObject;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+
+@Path("/test")
+@ApplicationScoped
+public class KafkaSaslResource {
+
+    @Path("/kafka/{topicName}")
+    @POST
+    @Produces(MediaType.APPLICATION_JSON)
+    public JsonObject post(@PathParam("topicName") String topicName, String message) throws Exception {
+        try (Producer<Integer, String> producer = KafkaSupport.createProducer()) {
+            RecordMetadata meta = producer.send(new ProducerRecord<>(topicName, 1, message)).get();
+
+            return Json.createObjectBuilder()
+                    .add("topicName", meta.topic())
+                    .add("partition", meta.partition())
+                    .add("offset", meta.offset())
+                    .build();
+        }
+    }
+
+    @Path("/kafka/{topicName}")
+    @GET
+    @Produces(MediaType.APPLICATION_JSON)
+    public JsonObject get(@PathParam("topicName") String topicName) {
+        try (KafkaConsumer<Integer, String> consumer = KafkaSupport.createConsumer(topicName)) {
+            ConsumerRecord<Integer, String> record = consumer.poll(Duration.ofSeconds(60)).iterator().next();
+            return Json.createObjectBuilder()
+                    .add("topicName", record.topic())
+                    .add("partition", record.partition())
+                    .add("offset", record.offset())
+                    .add("key", record.key())
+                    .add("body", record.value())
+                    .build();
+        }
+    }
+}

integration-tests/kafka-sasl/src/main/java/org/apache/camel/quarkus/kafka/sasl/KafkaSaslRoutes.java

@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.kafka.sasl;
+
+import org.apache.camel.builder.RouteBuilder;
+
+public class KafkaSaslRoutes extends RouteBuilder {
+    @Override
+    public void configure() throws Exception {
+        // Note: kafka component configuration is done via quarkus-kubernetes-service-binding.
+        // See configuration in src/test/resources/k8s-sb/kafka
+        from("kafka:inbound")
+                .to("log:kafka")
+                .to("kafka:outbound");
+    }
+}