Improve Kafka integration with Quarkus dev services

Fixes apache#3121

docs/modules/ROOT/pages/reference/extensions/kafka.adoc

@@ -39,6 +39,17 @@ Or add the coordinates to your existing project:
 
 Check the xref:user-guide/index.adoc[User guide] for more information about writing Camel Quarkus applications.
 
+== Usage
+
+=== Quarkus Kafka Dev Services
+
+Camel Quarkus Kafka can take advantage of https://quarkus.io/guides/kafka-dev-services[Quarkus Kafka Dev services] to simplify
+development and testing with a local containerized Kafka broker.
+Kafka Dev Services is enabled by default in dev & test mode. The Camel Kafka component is automatically configured so that the `brokers`
+component option is set to point at the local containerized Kafka broker. Meaning that there's no need to configure this option yourself.
+This functionality can be disabled with the configuration property `quarkus.kafka.devservices.enabled=false`.
+
+
 == Additional Camel Quarkus configuration
 
 [width="100%",cols="80,5,15",options="header"]

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

@@ -16,13 +16,26 @@
  */
 package org.apache.camel.quarkus.component.kafka.deployment;
 
+import java.util.Optional;
+
 import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
 import io.quarkus.deployment.Capabilities;
 import io.quarkus.deployment.Capability;
+import io.quarkus.deployment.IsNormal;
 import io.quarkus.deployment.annotations.BuildProducer;
 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.dev.devservices.GlobalDevServicesConfig;
+import io.quarkus.kafka.client.deployment.DevServicesKafkaBrokerBuildItem;
+import io.quarkus.kafka.client.deployment.KafkaBuildTimeConfig;
+import org.apache.camel.component.kafka.KafkaComponent;
+import org.apache.camel.quarkus.component.kafka.CamelKafkaRecorder;
 import org.apache.camel.quarkus.component.kafka.KafkaClientFactoryProducer;
+import org.apache.camel.quarkus.core.deployment.spi.CamelBeanBuildItem;
+import org.eclipse.microprofile.config.Config;
+import org.eclipse.microprofile.config.ConfigProvider;
 
 class KafkaProcessor {
     private static final String FEATURE = "camel-kafka";
@@ -40,4 +53,22 @@ void createKafkaClientFactoryProducerBean(
             additionalBean.produce(AdditionalBeanBuildItem.unremovableOf(KafkaClientFactoryProducer.class));
         }
     }
+
+    @Record(ExecutionTime.STATIC_INIT)
+    @BuildStep(onlyIfNot = IsNormal.class, onlyIf = GlobalDevServicesConfig.Enabled.class)
+    public void configureKafkaComponentForDevServices(
+            DevServicesKafkaBrokerBuildItem kafkaBrokerBuildItem,
+            KafkaBuildTimeConfig kafkaBuildTimeConfig,
+            BuildProducer<CamelBeanBuildItem> camelBean,
+            CamelKafkaRecorder recorder) {
+
+        Config config = ConfigProvider.getConfig();
+        Optional<String> brokers = config.getOptionalValue("camel.component.kafka.brokers", String.class);
+        if (brokers.isEmpty() && kafkaBuildTimeConfig.devservices.enabled.orElse(true)) {
+            camelBean.produce(new CamelBeanBuildItem(
+                    "kafka",
+                    KafkaComponent.class.getName(),
+                    recorder.createKafkaComponentForDevServices(kafkaBrokerBuildItem.getBootstrapServers())));
+        }
+    }
 }

extensions/kafka/deployment/src/test/java/org/apache/camel/quarkus/component/kafka/deployment/KafkaDevServicesDisabledTest.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.deployment;
+
+import javax.inject.Inject;
+
+import io.quarkus.test.QuarkusUnitTest;
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.kafka.KafkaComponent;
+import org.apache.camel.component.kafka.KafkaConfiguration;
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+public class KafkaDevServicesDisabledTest {
+
+    @RegisterExtension
+    static final QuarkusUnitTest CONFIG = new QuarkusUnitTest()
+            .withConfigurationResource("application-configuration-devservices-disabled.properties")
+            .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class));
+
+    @Inject
+    CamelContext context;
+
+    @Test
+    public void camelKafkaComponentBrokersConfigurationNull() {
+        KafkaComponent component = context.getComponent("kafka", KafkaComponent.class);
+        KafkaConfiguration configuration = component.getConfiguration();
+        assertNotNull(configuration);
+        assertNull(configuration.getBrokers());
+    }
+}

extensions/kafka/deployment/src/test/java/org/apache/camel/quarkus/component/kafka/deployment/KafkaDevServicesEnabledTest.java

@@ -0,0 +1,66 @@
+/*
+ * 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.deployment;
+
+import javax.inject.Inject;
+
+import io.quarkus.test.QuarkusUnitTest;
+import org.apache.camel.CamelContext;
+import org.apache.camel.ConsumerTemplate;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.kafka.KafkaComponent;
+import org.apache.camel.component.kafka.KafkaConfiguration;
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+public class KafkaDevServicesEnabledTest {
+
+    @RegisterExtension
+    static final QuarkusUnitTest CONFIG = new QuarkusUnitTest()
+            .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class));
+
+    @Inject
+    CamelContext context;
+
+    @Inject
+    ProducerTemplate producerTemplate;
+
+    @Inject
+    ConsumerTemplate consumerTemplate;
+
+    @ConfigProperty(name = "kafka.bootstrap.servers")
+    String bootstrapServers;
+
+    @Test
+    public void camelKafkaComponentBrokersConfigurationIsDevServicesBroker() {
+        KafkaComponent component = context.getComponent("kafka", KafkaComponent.class);
+        KafkaConfiguration configuration = component.getConfiguration();
+        assertNotNull(configuration);
+        assertEquals(bootstrapServers, configuration.getBrokers());
+
+        String payload = "Hello World";
+        producerTemplate.sendBody("kafka:test?autoOffsetReset=earliest", payload);
+        String result = consumerTemplate.receiveBody("kafka:test?autoOffsetReset=earliest", 5000, String.class);
+        assertEquals(payload, result);
+    }
+}

extensions/kafka/deployment/src/test/resources/application-configuration-devservices-disabled.properties

@@ -0,0 +1,17 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+quarkus.kafka.devservices.enabled=false

extensions/kafka/runtime/src/main/doc/usage.adoc

@@ -0,0 +1,9 @@
+=== Quarkus Kafka Dev Services
+
+Camel Quarkus Kafka can take advantage of https://quarkus.io/guides/kafka-dev-services[Quarkus Kafka Dev services] to simplify development and testing with a local containerized Kafka broker.
+
+Kafka Dev Services is enabled by default in dev & test mode.
+The Camel Kafka component is automatically configured so that the `brokers` component option is set to point at the local containerized Kafka broker.
+Meaning that there's no need to configure this option yourself.
+
+This functionality can be disabled with the configuration property `quarkus.kafka.devservices.enabled=false`.

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

@@ -0,0 +1,36 @@
+/*
+ * 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 io.quarkus.runtime.RuntimeValue;
+import io.quarkus.runtime.annotations.Recorder;
+import org.apache.camel.component.kafka.KafkaComponent;
+import org.apache.camel.component.kafka.KafkaConfiguration;
+
+@Recorder
+public class CamelKafkaRecorder {
+
+    public RuntimeValue<KafkaComponent> createKafkaComponentForDevServices(String brokers) {
+        KafkaConfiguration configuration = new KafkaConfiguration();
+        configuration.setBrokers(brokers);
+
+        KafkaComponent component = new KafkaComponent();
+        component.setConfiguration(configuration);
+
+        return new RuntimeValue<>(component);
+    }
+}