From bdcda0caf346666ad083deb006fdb1c568037c6d Mon Sep 17 00:00:00 2001 From: Guillaume Nodet Date: Thu, 30 Jul 2026 08:39:29 +0200 Subject: [PATCH] CAMEL-24305: Fix autowiring of KafkaClientFactory via addComponent() Move default KafkaClientFactory creation from doInit() to doStart() in KafkaComponent. When a component is registered via addComponent() (the path used by Spring Boot), doInit() runs before the autowiring lifecycle strategy, causing the default factory to block injection of a custom one. Also restore the DefaultKafkaClientFactory fallback in KafkaEndpoint's doBuild() method for resilience when the endpoint builds before the component starts. Co-Authored-By: Claude Opus 4.6 --- .../camel/component/kafka/KafkaComponent.java | 12 ++--- .../camel/component/kafka/KafkaEndpoint.java | 3 ++ .../component/kafka/KafkaAutowireTest.java | 49 ++++++++++++++++++- 3 files changed, 55 insertions(+), 9 deletions(-) diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java index 3883d192bb20b..551105bbff68a 100644 --- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java +++ b/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 map = new HashMap<>(); // resolve parameter values from the values (#bean / #class etc) diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java index 41549edbba451..30746c1de1731 100644 --- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java +++ b/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(); } diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaAutowireTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaAutowireTest.java index d96842acab986..1493d3f96c89c 100644 --- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaAutowireTest.java +++ b/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,7 +38,7 @@ 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()); @@ -44,6 +46,49 @@ public void testKafkaComponentAutowiring() { 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 CAMEL-24305. + */ + @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 { }