CAMEL-24305: Fix autowiring of KafkaClientFactory via addComponent()

components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java

@@ -298,21 +298,19 @@ private void startPendingConsumers() {
     }
 
     @Override
-    protected void doInit() throws Exception {
-        super.doInit();
+    protected void doStart() throws Exception {
+        super.doStart();
 
         // if a factory was not autowired then create a default factory
+        // NOTE: must be done in doStart() rather than doInit(), because when a component is
+        // registered via addComponent() (the path used by Spring Boot), doInit() runs before
+        // the autowiring lifecycle strategy has a chance to inject a custom factory.
         if (kafkaClientFactory == null) {
             kafkaClientFactory = new DefaultKafkaClientFactory();
         }
         if (configuration.isAllowManualCommit() && kafkaManualCommitFactory == null) {
             LOG.warn("The component was setup for allowing manual commits, but a manual commit factory was not set");
         }
-    }
-
-    @Override
-    protected void doStart() throws Exception {
-        super.doStart();
 
         Map<String, Object> map = new HashMap<>();
         // resolve parameter values from the values (#bean / #class etc)

components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java

@@ -132,6 +132,9 @@ protected void doBuild() throws Exception {
         if (kafkaClientFactory == null) {
             kafkaClientFactory = getComponent().getKafkaClientFactory();
         }
+        if (kafkaClientFactory == null) {
+            kafkaClientFactory = new DefaultKafkaClientFactory();
+        }
         if (kafkaManualCommitFactory == null) {
             kafkaManualCommitFactory = getComponent().getKafkaManualCommitFactory();
         }

components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaAutowireTest.java

@@ -18,14 +18,16 @@
 
 import org.apache.camel.BindToRegistry;
 import org.apache.camel.CamelContext;
+import org.apache.camel.impl.DefaultCamelContext;
 import org.apache.camel.test.infra.core.CamelContextExtension;
 import org.apache.camel.test.infra.core.DefaultCamelContextExtension;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertSame;
 
-public class KafkaAutowireTest {
+class KafkaAutowireTest {
 
     @RegisterExtension
     protected static CamelContextExtension contextExtension = new DefaultCamelContextExtension();
@@ -36,14 +38,57 @@ public class KafkaAutowireTest {
     private final KafkaClientFactory clientFactory = new TestKafkaClientFactory();
 
     @Test
-    public void testKafkaComponentAutowiring() {
+    void testKafkaComponentAutowiring() {
         KafkaComponent component = context.getComponent("kafka", KafkaComponent.class);
         assertSame(clientFactory, component.getKafkaClientFactory());
 
         KafkaEndpoint endpoint = context.getEndpoint("kafka:foo", KafkaEndpoint.class);
         assertSame(clientFactory, endpoint.getKafkaClientFactory());
     }
 
+    /**
+     * Verifies that autowiring works when the component is registered via
+     * {@link CamelContext#addComponent(String, org.apache.camel.Component)}, which is the path used by Spring Boot. In
+     * this path, the component is added before the context is started, so {@code doInit()} runs before the autowiring
+     * lifecycle strategy. The default factory must not be created until {@code doStart()} to give autowiring a chance
+     * to inject a custom factory.
+     *
+     * This is a regression test for <a href="https://issues.apache.org/jira/browse/CAMEL-24305">CAMEL-24305</a>.
+     */
+    @Test
+    void testKafkaComponentAutowiringViaAddComponent() throws Exception {
+        // Simulate the Spring Boot path: addComponent() is called before the context is started
+        try (DefaultCamelContext ctx = new DefaultCamelContext()) {
+            KafkaClientFactory customFactory = new TestKafkaClientFactory();
+            ctx.getRegistry().bind("kafkaClientFactory", customFactory);
+
+            KafkaComponent component = new KafkaComponent();
+            ctx.addComponent("kafka", component);
+
+            ctx.start();
+
+            assertSame(customFactory, component.getKafkaClientFactory(),
+                    "Custom KafkaClientFactory should be autowired when using addComponent()");
+        }
+    }
+
+    /**
+     * Verifies that when no custom KafkaClientFactory is registered, the component creates a
+     * {@link DefaultKafkaClientFactory} as the default.
+     */
+    @Test
+    void testKafkaComponentDefaultFactoryWhenNoneRegistered() throws Exception {
+        try (DefaultCamelContext ctx = new DefaultCamelContext()) {
+            KafkaComponent component = new KafkaComponent();
+            ctx.addComponent("kafka", component);
+
+            ctx.start();
+
+            assertInstanceOf(DefaultKafkaClientFactory.class, component.getKafkaClientFactory(),
+                    "Default KafkaClientFactory should be created when none is registered");
+        }
+    }
+
     static final class TestKafkaClientFactory extends DefaultKafkaClientFactory {
 
     }