You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ja...@apache.org on 2026/04/17 08:33:35 UTC
(camel-quarkus) 03/03: Ensure forward compatibility with Kafka 4.2.0
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 <oz...@gmail.com>
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