This is an automated email from the ASF dual-hosted git repository. jamesnetherton pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel-quarkus.git
commit 7a82a61879968bb22819da621cb4088e79427006 Author: Ozan Gunalp <[email protected]> AuthorDate: Thu Apr 16 02:30:09 2026 +0200 Ensure forward compatibility with Kafka 4.2.0 Add a DebeziumComponentObserver CDI bean that sets bootstrap.servers into Debezium component additional properties when the component is added to the CamelContext --- extensions-support/debezium/deployment/pom.xml | 4 +++ .../deployment/DebeziumSupportProcessor.java | 7 +++++ extensions-support/debezium/runtime/pom.xml | 4 +++ .../debezium/DebeziumComponentObserver.java | 35 ++++++++++++++++++++++ .../common/it/postgres/DebeziumPostgresTest.java | 3 +- 5 files changed, 52 insertions(+), 1 deletion(-) diff --git a/extensions-support/debezium/deployment/pom.xml b/extensions-support/debezium/deployment/pom.xml index 29eadd0030..57b139a59e 100644 --- a/extensions-support/debezium/deployment/pom.xml +++ b/extensions-support/debezium/deployment/pom.xml @@ -38,6 +38,10 @@ <groupId>io.quarkus</groupId> <artifactId>quarkus-datasource-deployment</artifactId> </dependency> + <dependency> + <groupId>org.apache.camel.quarkus</groupId> + <artifactId>camel-quarkus-core-deployment</artifactId> + </dependency> <dependency> <groupId>org.apache.camel.quarkus</groupId> <artifactId>camel-quarkus-support-debezium</artifactId> diff --git a/extensions-support/debezium/deployment/src/main/java/org/apache/camel/quarkus/support/debezium/deployment/DebeziumSupportProcessor.java b/extensions-support/debezium/deployment/src/main/java/org/apache/camel/quarkus/support/debezium/deployment/DebeziumSupportProcessor.java index ff73bf58e0..d630f38bd8 100644 --- a/extensions-support/debezium/deployment/src/main/java/org/apache/camel/quarkus/support/debezium/deployment/DebeziumSupportProcessor.java +++ b/extensions-support/debezium/deployment/src/main/java/org/apache/camel/quarkus/support/debezium/deployment/DebeziumSupportProcessor.java @@ -41,6 +41,7 @@ import io.debezium.snapshot.mode.RecoverySnapshotter; import io.debezium.snapshot.mode.WhenNeededSnapshotter; import io.debezium.snapshot.spi.SnapshotLock; import io.debezium.storage.file.history.FileSchemaHistory; +import io.quarkus.arc.deployment.AdditionalBeanBuildItem; import io.quarkus.deployment.annotations.BuildProducer; import io.quarkus.deployment.annotations.BuildStep; import io.quarkus.deployment.builditem.CombinedIndexBuildItem; @@ -48,6 +49,7 @@ import io.quarkus.deployment.builditem.IndexDependencyBuildItem; import io.quarkus.deployment.builditem.nativeimage.NativeImageResourceBuildItem; import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem; import io.quarkus.deployment.builditem.nativeimage.RuntimeInitializedClassBuildItem; +import org.apache.camel.quarkus.support.debezium.DebeziumComponentObserver; import org.apache.kafka.common.security.authenticator.SaslClientAuthenticator; import org.apache.kafka.connect.json.JsonConverter; import org.apache.kafka.connect.source.SourceTask; @@ -67,6 +69,11 @@ public class DebeziumSupportProcessor { return new RuntimeInitializedClassBuildItem(SaslClientAuthenticator.class.getName()); } + @BuildStep + public void configureKafkaComponentForDevServices(BuildProducer<AdditionalBeanBuildItem> additionalBean) { + additionalBean.produce(AdditionalBeanBuildItem.unremovableOf(DebeziumComponentObserver.class)); + } + @BuildStep void reflectiveClasses(CombinedIndexBuildItem combinedIndex, BuildProducer<ReflectiveClassBuildItem> reflectiveClasses) { IndexView index = combinedIndex.getIndex(); diff --git a/extensions-support/debezium/runtime/pom.xml b/extensions-support/debezium/runtime/pom.xml index 0b93fefe5e..d3c1a692c6 100644 --- a/extensions-support/debezium/runtime/pom.xml +++ b/extensions-support/debezium/runtime/pom.xml @@ -47,6 +47,10 @@ <groupId>org.javassist</groupId> <artifactId>javassist</artifactId> </dependency> + <dependency> + <groupId>org.apache.camel.quarkus</groupId> + <artifactId>camel-quarkus-core</artifactId> + </dependency> <dependency> <groupId>org.apache.camel</groupId> <artifactId>camel-debezium-common</artifactId> diff --git a/extensions-support/debezium/runtime/src/main/java/org/apache/camel/quarkus/support/debezium/DebeziumComponentObserver.java b/extensions-support/debezium/runtime/src/main/java/org/apache/camel/quarkus/support/debezium/DebeziumComponentObserver.java new file mode 100644 index 0000000000..a84c2456a6 --- /dev/null +++ b/extensions-support/debezium/runtime/src/main/java/org/apache/camel/quarkus/support/debezium/DebeziumComponentObserver.java @@ -0,0 +1,35 @@ +/* + * 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.support.debezium; + +import java.util.Map; + +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.enterprise.event.Observes; +import org.apache.camel.component.debezium.DebeziumComponent; +import org.apache.camel.quarkus.core.events.ComponentAddEvent; + +@ApplicationScoped +public class DebeziumComponentObserver { + + void initDebeziumBootstrapServers(@Observes ComponentAddEvent event) { + if (event.getComponent() instanceof DebeziumComponent<?> debeziumComponent) { + Map<String, Object> additionalProperties = debeziumComponent.getConfiguration().getAdditionalProperties(); + additionalProperties.putIfAbsent("bootstrap.servers", "localhost:9092"); + } + } +} diff --git a/integration-test-groups/debezium/postgresql/src/test/java/org/apache/camel/quarkus/component/debezium/common/it/postgres/DebeziumPostgresTest.java b/integration-test-groups/debezium/postgresql/src/test/java/org/apache/camel/quarkus/component/debezium/common/it/postgres/DebeziumPostgresTest.java index dd92f423bc..5732de2103 100644 --- a/integration-test-groups/debezium/postgresql/src/test/java/org/apache/camel/quarkus/component/debezium/common/it/postgres/DebeziumPostgresTest.java +++ b/integration-test-groups/debezium/postgresql/src/test/java/org/apache/camel/quarkus/component/debezium/common/it/postgres/DebeziumPostgresTest.java @@ -58,7 +58,8 @@ class DebeziumPostgresTest extends AbstractDebeziumTest { RestAssured.get(Type.postgres.getComponent() + "/getAdditionalProperties") .then() .statusCode(200) - .body("'database.connectionTimeZone'", is("CET")); + .body("'database.connectionTimeZone'", is("CET")) + .body("'bootstrap.servers'", is(notNullValue())); } @AfterAll
