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
The following commit(s) were added to refs/heads/main by this push:
new 42cccd42eec Fixes #9261. Make Debezium extensions more resilient to
version upgrades
42cccd42eec is described below
commit 42cccd42eecc9a16e3379d25d0799691dd20af10
Author: James Netherton <[email protected]>
AuthorDate: Mon Nov 2 16:17:06 2026 +0000
Fixes #9261. Make Debezium extensions more resilient to version upgrades
Replace the hand-maintained native registrations with build-time discovery:
* Index the io.debezium runtime dependencies (except debezium-ddl-parser)
and
register providers for all META-INF/services interfaces they declare,
excluding services the embedded engine never loads (component metadata,
sink connectors, transformations, Hibernate, Infinispan). Converter and
HeaderConverter providers are always registered
* Register implementations of the plugin types Debezium instantiates from
configured class names (SchemaHistory, SourceInfoStructMaker, SourceTask,
TopicNamingStrategy, TransactionMetadataFactory, TableMappingStorage,
ReselectColumnCache)
* Reduce the connector processors to their connector-specific registrations
* Remove registrations for classes and resources that no longer exist
* Add a test that fails when a Debezium configuration option defaults to a
class that is not registered for reflection
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
extensions-support/debezium/deployment/pom.xml | 30 ++++
.../deployment/DebeziumSupportProcessor.java | 197 ++++++++++-----------
.../DebeziumConfigurablePluginsTest.java | 143 +++++++++++++++
.../deployment/DebeziumMongodbProcessor.java | 29 ++-
.../mysql/deployment/DebeziumMysqlProcessor.java | 23 ---
.../oracle/deployment/DebeziumOracleProcessor.java | 19 +-
.../deployment/DebeziumPostgresProcessor.java | 20 ---
.../deployment/DebeziumSqlserverProcessor.java | 27 ---
8 files changed, 282 insertions(+), 206 deletions(-)
diff --git a/extensions-support/debezium/deployment/pom.xml
b/extensions-support/debezium/deployment/pom.xml
index b831b15d61f..a155030193c 100644
--- a/extensions-support/debezium/deployment/pom.xml
+++ b/extensions-support/debezium/deployment/pom.xml
@@ -50,6 +50,36 @@
<groupId>org.apache.camel.quarkus</groupId>
<artifactId>camel-quarkus-support-debezium</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-debezium-mongodb</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-debezium-mysql</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-debezium-oracle</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-debezium-postgres</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-debezium-sqlserver</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.quarkus</groupId>
+ <artifactId>quarkus-junit-internal</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
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 dd4cf816e13..96b99d9b933 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
@@ -16,52 +16,84 @@
*/
package org.apache.camel.quarkus.support.debezium.deployment;
-import io.debezium.connector.base.DefaultQueueProvider;
-import io.debezium.connector.common.BaseSourceTask;
-import io.debezium.embedded.async.ConvertingAsyncEngineBuilderFactory;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import io.debezium.connector.SourceInfoStructMaker;
import io.debezium.engine.DebeziumEngine;
-import io.debezium.pipeline.notification.channels.NotificationChannel;
-import io.debezium.pipeline.signal.actions.SignalActionProvider;
-import io.debezium.pipeline.signal.actions.StandardActionProvider;
-import io.debezium.pipeline.signal.channels.FileSignalChannel;
-import io.debezium.pipeline.signal.channels.KafkaSignalChannel;
-import io.debezium.pipeline.signal.channels.SourceSignalChannel;
-import io.debezium.pipeline.signal.channels.jmx.JmxSignalChannel;
-import io.debezium.pipeline.signal.channels.process.InProcessSignalChannel;
-import io.debezium.pipeline.txmetadata.DefaultTransactionMetadataFactory;
-import io.debezium.schema.SchemaTopicNamingStrategy;
-import io.debezium.snapshot.lock.NoLockingSupport;
-import io.debezium.snapshot.mode.AlwaysSnapshotter;
-import io.debezium.snapshot.mode.ConfigurationBasedSnapshotter;
-import io.debezium.snapshot.mode.InitialOnlySnapshotter;
-import io.debezium.snapshot.mode.InitialSnapshotter;
-import io.debezium.snapshot.mode.NoDataSnapshotter;
-import io.debezium.snapshot.mode.RecoverySnapshotter;
-import io.debezium.snapshot.mode.WhenNeededNoDataSnapshotter;
-import io.debezium.snapshot.mode.WhenNeededSnapshotter;
+import io.debezium.pipeline.txmetadata.spi.TransactionMetadataFactory;
+import io.debezium.processors.PostProcessorRegistry;
+import io.debezium.processors.reselect.cache.ReselectColumnCache;
+import io.debezium.relational.TableMappingStorage;
+import io.debezium.relational.history.SchemaHistory;
import io.debezium.snapshot.spi.SnapshotLock;
-import io.debezium.spi.storage.OffsetStoreProvider;
-import io.debezium.storage.file.history.FileSchemaHistory;
+import io.debezium.spi.topic.TopicNamingStrategy;
import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
import io.quarkus.deployment.annotations.BuildProducer;
import io.quarkus.deployment.annotations.BuildStep;
import io.quarkus.deployment.builditem.CombinedIndexBuildItem;
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.ServiceProviderBuildItem;
+import io.quarkus.deployment.pkg.builditem.CurateOutcomeBuildItem;
+import io.quarkus.maven.dependency.ResolvedDependency;
+import io.quarkus.paths.PathFilter;
import
org.apache.camel.quarkus.core.deployment.spi.CamelSerializationBuildItem;
import org.apache.camel.quarkus.support.debezium.DebeziumComponentObserver;
import org.apache.kafka.connect.source.SourceTask;
+import org.apache.kafka.connect.storage.Converter;
+import org.apache.kafka.connect.storage.FileOffsetBackingStore;
+import org.apache.kafka.connect.storage.HeaderConverter;
+import org.apache.kafka.connect.storage.KafkaOffsetBackingStore;
+import org.apache.kafka.connect.storage.MemoryOffsetBackingStore;
+import org.jboss.jandex.ClassInfo;
import org.jboss.jandex.DotName;
import org.jboss.jandex.IndexView;
public class DebeziumSupportProcessor {
+ private static final String DEBEZIUM_GROUP_ID = "io.debezium";
+ private static final String SERVICES_PATH = "META-INF/services/";
+ // Holds no plugin implementations and would add thousands of ANTLR
classes to the index
+ private static final Set<String> UNINDEXED_ARTIFACTS =
Set.of("debezium-ddl-parser");
+
+ // Types whose implementations Debezium instantiates from class names held
in connector configuration
+ static final List<Class<?>> CONFIGURABLE_PLUGIN_TYPES = List.of(
+ ReselectColumnCache.class,
+ SchemaHistory.class,
+ SourceInfoStructMaker.class,
+ SourceTask.class,
+ TableMappingStorage.class,
+ TopicNamingStrategy.class,
+ TransactionMetadataFactory.class);
+
+ // Classes named by configuration defaults that are registered for
reflection individually
+ static final List<Class<?>> REFLECTIVE_CLASSES = List.of(
+ FileOffsetBackingStore.class,
+ MemoryOffsetBackingStore.class,
+ KafkaOffsetBackingStore.class,
+ DebeziumEngine.BuilderFactory.class,
+ SnapshotLock.class,
+ SourceTask.class);
+
+ // Services registered whether or not a Debezium artifact declares them
+ static final List<Class<?>> REQUIRED_SERVICES = List.of(Converter.class,
HeaderConverter.class);
+
+ // Services declared by Debezium artifacts that the embedded engine never
loads through ServiceLoader
+ static final Set<String> EXCLUDED_SERVICES = Set.of(
+ "io.debezium.metadata.ComponentMetadataProvider",
+ "org.apache.kafka.connect.sink.SinkConnector",
+ "org.apache.kafka.connect.transforms.Transformation",
+ "org.hibernate.boot.spi.MetadataSourcesContributor",
+ "org.infinispan.protostream.SerializationContextInitializer");
@BuildStep
- void addDependencies(BuildProducer<IndexDependencyBuildItem>
indexDependency) {
+ void addDependencies(CurateOutcomeBuildItem curateOutcome,
BuildProducer<IndexDependencyBuildItem> indexDependency) {
indexDependency.produce(new
IndexDependencyBuildItem("org.apache.kafka", "connect-json"));
- indexDependency.produce(new IndexDependencyBuildItem("io.debezium",
"debezium-api"));
+ debeziumDependencies(curateOutcome).stream()
+ .filter(dependency ->
!UNINDEXED_ARTIFACTS.contains(dependency.getArtifactId()))
+ .forEach(dependency -> indexDependency
+ .produce(new
IndexDependencyBuildItem(dependency.getGroupId(), dependency.getArtifactId())));
}
@BuildStep
@@ -95,85 +127,52 @@ public class DebeziumSupportProcessor {
.toArray(String[]::new);
reflectiveClasses.produce(ReflectiveClassBuildItem.builder(dtos).fields().build());
- reflectiveClasses.produce(ReflectiveClassBuildItem.builder(
- "org.apache.kafka.connect.storage.FileOffsetBackingStore",
- "org.apache.kafka.connect.storage.MemoryOffsetBackingStore",
- "org.apache.kafka.connect.storage.KafkaOffsetBackingStore",
- "io.debezium.storage.kafka.history.KafkaSchemaHistory",
- "io.debezium.relational.history.FileDatabaseHistory",
- "io.debezium.embedded.ConvertingEngineBuilderFactory",
- "io.debezium.relational.ConcurrentMapTableMappingStorage")
- .build());
+ String[] plugins = CONFIGURABLE_PLUGIN_TYPES.stream()
+ .flatMap(type -> {
+ DotName name = DotName.createSimple(type);
+ return type.isInterface()
+ ? index.getAllKnownImplementations(name).stream()
+ : index.getAllKnownSubclasses(name).stream();
+ })
+ .filter(ci -> !ci.isInterface())
+ .map(ClassInfo::name)
+ .map(DotName::toString)
+ .toArray(String[]::new);
+
reflectiveClasses.produce(ReflectiveClassBuildItem.builder(plugins).build());
-
reflectiveClasses.produce(ReflectiveClassBuildItem.builder("io.debezium.processors.PostProcessorRegistry")
- .methods()
- .build());
+
reflectiveClasses.produce(ReflectiveClassBuildItem.builder(REFLECTIVE_CLASSES.toArray(Class<?>[]::new)).build());
- reflectiveClasses.produce(ReflectiveClassBuildItem.builder(
- DebeziumEngine.BuilderFactory.class,
- ConvertingAsyncEngineBuilderFactory.class,
- DefaultTransactionMetadataFactory.class,
- SchemaTopicNamingStrategy.class,
- BaseSourceTask.class,
- SnapshotLock.class,
- NoLockingSupport.class,
- AlwaysSnapshotter.class,
- InitialSnapshotter.class,
- InitialOnlySnapshotter.class,
- NoDataSnapshotter.class,
- RecoverySnapshotter.class,
- WhenNeededSnapshotter.class,
- WhenNeededNoDataSnapshotter.class,
- ConfigurationBasedSnapshotter.class,
- SourceSignalChannel.class,
- KafkaSignalChannel.class,
- FileSignalChannel.class,
- JmxSignalChannel.class,
- InProcessSignalChannel.class,
- StandardActionProvider.class,
- SourceTask.class,
- FileSchemaHistory.class,
- DefaultQueueProvider.class)
+
reflectiveClasses.produce(ReflectiveClassBuildItem.builder(PostProcessorRegistry.class)
+ .methods()
.build());
-
- }
-
- @BuildStep
- ServiceProviderBuildItem registerConverterServiceProviders() {
- return
ServiceProviderBuildItem.allProvidersFromClassPath("org.apache.kafka.connect.storage.Converter");
}
@BuildStep
- ServiceProviderBuildItem registerSignalActionProviders() {
- return
ServiceProviderBuildItem.allProvidersFromClassPath(SignalActionProvider.class.getName());
+ void registerServiceProviders(
+ CurateOutcomeBuildItem curateOutcome,
+ BuildProducer<ServiceProviderBuildItem> serviceProvider) {
+ Set<String> services = new HashSet<>();
+ REQUIRED_SERVICES.forEach(service -> services.add(service.getName()));
+ PathFilter pathFilter = PathFilter.forIncludes(Set.of(SERVICES_PATH +
"*"));
+ debeziumDependencies(curateOutcome)
+ .forEach(dependency ->
dependency.getContentTree(pathFilter).walk(visit -> {
+ String resourceName = visit.getResourceName();
+ if (resourceName.length() > SERVICES_PATH.length()) {
+
services.add(resourceName.substring(SERVICES_PATH.length()));
+ }
+ }));
+
+ services.stream()
+ .filter(service -> !EXCLUDED_SERVICES.contains(service))
+ .map(ServiceProviderBuildItem::allProvidersFromClassPath)
+ .forEach(serviceProvider::produce);
}
- @BuildStep
- ServiceProviderBuildItem registerNotificationChannelProviders() {
- return
ServiceProviderBuildItem.allProvidersFromClassPath(NotificationChannel.class.getName());
- }
-
- @BuildStep
- ServiceProviderBuildItem registerOffsetStoreProviders() {
- return
ServiceProviderBuildItem.allProvidersFromClassPath(OffsetStoreProvider.class.getName());
+ private static List<ResolvedDependency>
debeziumDependencies(CurateOutcomeBuildItem curateOutcome) {
+ return curateOutcome.getApplicationModel()
+ .getRuntimeDependencies()
+ .stream()
+ .filter(dependency ->
dependency.getGroupId().equals(DEBEZIUM_GROUP_ID))
+ .toList();
}
-
- @BuildStep
- void
registerNativeImageResources(BuildProducer<NativeImageResourceBuildItem>
resources) {
- resources.produce(new NativeImageResourceBuildItem(
-
"META-INF/services/io.debezium.embedded.async.ConvertingAsyncEngineBuilderFactory"));
- resources.produce(
- new
NativeImageResourceBuildItem("META-INF/services/io.debezium.engine.DebeziumEngine$BuilderFactory"));
- resources.produce(new
NativeImageResourceBuildItem("META-INF/services/io.debezium.spi.snapshot.Snapshotter"));
- resources.produce(
- new
NativeImageResourceBuildItem("META-INF/services/io.debezium.pipeline.signal.channels.SignalChannelReader"));
- resources.produce(new
NativeImageResourceBuildItem("META-INF/services/io.debezium.processors.PostProcessorProducer"));
-
- resources.produce(new
NativeImageResourceBuildItem("META-INF/services/io.debezium.snapshot.spi.SnapshotLock"));
- resources.produce(new
NativeImageResourceBuildItem("META-INF/services/io.debezium.snapshot.spi.SnapshotQuery"));
- resources.produce(new
NativeImageResourceBuildItem("META-INF/services/io.debezium.connector.base.QueueProvider"));
- resources
- .produce(new
NativeImageResourceBuildItem("META-INF/services/org.apache.kafka.connect.source.SourceConnector"));
- }
-
}
diff --git
a/extensions-support/debezium/deployment/src/test/java/org/apache/camel/quarkus/support/debezium/deployment/DebeziumConfigurablePluginsTest.java
b/extensions-support/debezium/deployment/src/test/java/org/apache/camel/quarkus/support/debezium/deployment/DebeziumConfigurablePluginsTest.java
new file mode 100644
index 00000000000..6981edaf6c4
--- /dev/null
+++
b/extensions-support/debezium/deployment/src/test/java/org/apache/camel/quarkus/support/debezium/deployment/DebeziumConfigurablePluginsTest.java
@@ -0,0 +1,143 @@
+/*
+ * 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.deployment;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.lang.reflect.Modifier;
+import java.net.JarURLConnection;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.jar.JarEntry;
+import java.util.jar.JarFile;
+
+import io.debezium.config.Field;
+import org.apache.kafka.common.config.ConfigDef;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Fails when a Debezium upgrade adds a configuration option whose default
value names a class that
+ * {@link DebeziumSupportProcessor} does not register for reflection. Only
options of type
+ * {@link ConfigDef.Type#CLASS} are checked: class names held in options of
other types, such as
+ * {@link ConfigDef.Type#STRING}, are not covered.
+ */
+class DebeziumConfigurablePluginsTest {
+ private static final String SERVICES_PATH = "META-INF/services/";
+ // Sink connectors are not supported by the Camel Debezium components
+ private static final String SINK_PACKAGE = ".sink.";
+
+ @Test
+ void configurationClassDefaultsAreRegisteredForReflection() throws
Exception {
+ ClassLoader classLoader =
Thread.currentThread().getContextClassLoader();
+ Set<String> serviceProviders = new HashSet<>();
+ Set<String> checked = new HashSet<>();
+ Set<String> unregistered = new TreeSet<>();
+
+ for (JarFile jar : debeziumJars(classLoader)) {
+ try (jar) {
+ for (JarEntry entry : Collections.list(jar.entries())) {
+ String entryName = entry.getName();
+ if (entryName.startsWith(SERVICES_PATH) &&
entryName.length() > SERVICES_PATH.length()
+ && !DebeziumSupportProcessor.EXCLUDED_SERVICES
+
.contains(entryName.substring(SERVICES_PATH.length()))) {
+ serviceProviders.addAll(readServiceProviders(jar,
entry));
+ } else if (entryName.endsWith(".class") &&
!entryName.contains("-")
+ && !entryName.replace('/',
'.').contains(SINK_PACKAGE)) {
+ String className = entryName.substring(0,
entryName.length() - 6).replace('/', '.');
+ for (String defaultClassName :
classDefaults(className, classLoader)) {
+ checked.add(defaultClassName);
+ if (!isRegistered(defaultClassName, classLoader)) {
+ unregistered.add(className + " -> " +
defaultClassName);
+ }
+ }
+ }
+ }
+ }
+ }
+
+ assertFalse(checked.isEmpty(), "No Debezium configuration options with
class defaults were found");
+ unregistered.removeIf(entry ->
serviceProviders.contains(entry.substring(entry.indexOf(" -> ") + 4)));
+ assertTrue(unregistered.isEmpty(),
+ "Debezium configuration defaults name classes that are not
registered for reflection. "
+ + "Add their plugin type to
DebeziumSupportProcessor.CONFIGURABLE_PLUGIN_TYPES "
+ + "or register them individually: " + unregistered);
+ }
+
+ private static Set<JarFile> debeziumJars(ClassLoader classLoader) throws
IOException, URISyntaxException {
+ Set<JarFile> jars = new HashSet<>();
+ for (URL url :
Collections.list(classLoader.getResources("META-INF/maven/io.debezium"))) {
+ if (url.openConnection() instanceof JarURLConnection connection) {
+ jars.add(new
JarFile(Path.of(connection.getJarFileURL().toURI()).toFile()));
+ }
+ }
+ return jars;
+ }
+
+ private static Set<String> readServiceProviders(JarFile jar, JarEntry
entry) throws IOException {
+ Set<String> providers = new HashSet<>();
+ try (BufferedReader reader = new BufferedReader(
+ new InputStreamReader(jar.getInputStream(entry),
StandardCharsets.UTF_8))) {
+ reader.lines()
+ .map(line -> line.replaceAll("#.*", "").trim())
+ .filter(line -> !line.isEmpty())
+ .forEach(providers::add);
+ }
+ return providers;
+ }
+
+ private static Set<String> classDefaults(String className, ClassLoader
classLoader) {
+ Set<String> defaults = new HashSet<>();
+ try {
+ Class<?> clazz = Class.forName(className, false, classLoader);
+ for (java.lang.reflect.Field field : clazz.getDeclaredFields()) {
+ if (Modifier.isStatic(field.getModifiers()) && field.getType()
== Field.class) {
+ field.setAccessible(true);
+ Field configField = (Field) field.get(null);
+ if (configField != null && configField.type() ==
ConfigDef.Type.CLASS
+ && configField.defaultValue() != null) {
+ Object defaultValue = configField.defaultValue();
+ defaults.add(defaultValue instanceof Class<?> type ?
type.getName() : defaultValue.toString());
+ }
+ }
+ }
+ } catch (Throwable e) {
+ // Classes that cannot be loaded or initialized without optional
dependencies hold no usable defaults
+ }
+ return defaults;
+ }
+
+ private static boolean isRegistered(String className, ClassLoader
classLoader) {
+ try {
+ Class<?> clazz = Class.forName(className, false, classLoader);
+ return DebeziumSupportProcessor.REFLECTIVE_CLASSES.contains(clazz)
+ ||
DebeziumSupportProcessor.CONFIGURABLE_PLUGIN_TYPES.stream()
+ .anyMatch(type -> type.isAssignableFrom(clazz));
+ } catch (ClassNotFoundException e) {
+ return false;
+ }
+ }
+}
diff --git
a/extensions/debezium-mongodb/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/mongodb/deployment/DebeziumMongodbProcessor.java
b/extensions/debezium-mongodb/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/mongodb/deployment/DebeziumMongodbProcessor.java
index ca9b9fac08c..653259cf6ca 100644
---
a/extensions/debezium-mongodb/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/mongodb/deployment/DebeziumMongodbProcessor.java
+++
b/extensions/debezium-mongodb/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/mongodb/deployment/DebeziumMongodbProcessor.java
@@ -16,16 +16,13 @@
*/
package org.apache.camel.quarkus.component.debezium.mongodb.deployment;
-import io.debezium.connector.mongodb.MongoDbConnector;
-import io.debezium.connector.mongodb.MongoDbConnectorTask;
-import io.debezium.connector.mongodb.MongoDbSourceInfoStructMaker;
-import io.debezium.connector.mongodb.connection.DefaultMongoDbAuthProvider;
-import io.debezium.connector.mongodb.snapshot.query.SelectAllSnapshotQuery;
+import io.debezium.connector.mongodb.connection.MongoDbAuthProvider;
import io.quarkus.deployment.annotations.BuildProducer;
import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.builditem.CombinedIndexBuildItem;
import io.quarkus.deployment.builditem.FeatureBuildItem;
-import io.quarkus.deployment.builditem.IndexDependencyBuildItem;
import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
+import org.jboss.jandex.DotName;
class DebeziumMongodbProcessor {
@@ -37,18 +34,12 @@ class DebeziumMongodbProcessor {
}
@BuildStep
- void reflectiveClasses(BuildProducer<ReflectiveClassBuildItem>
reflectiveClasses) {
-
- reflectiveClasses.produce(ReflectiveClassBuildItem.builder(
- MongoDbConnector.class,
- MongoDbConnectorTask.class,
- MongoDbSourceInfoStructMaker.class,
- DefaultMongoDbAuthProvider.class,
- SelectAllSnapshotQuery.class).build());
- }
-
- @BuildStep
- void addDependencies(BuildProducer<IndexDependencyBuildItem>
indexDependency) {
- indexDependency.produce(new IndexDependencyBuildItem("io.debezium",
"debezium-connector-mongodb"));
+ void reflectiveClasses(CombinedIndexBuildItem combinedIndex,
BuildProducer<ReflectiveClassBuildItem> reflectiveClasses) {
+ String[] authProviders = combinedIndex.getIndex()
+
.getAllKnownImplementations(DotName.createSimple(MongoDbAuthProvider.class))
+ .stream()
+ .map(ci -> ci.name().toString())
+ .toArray(String[]::new);
+
reflectiveClasses.produce(ReflectiveClassBuildItem.builder(authProviders).build());
}
}
diff --git
a/extensions/debezium-mysql/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/mysql/deployment/DebeziumMysqlProcessor.java
b/extensions/debezium-mysql/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/mysql/deployment/DebeziumMysqlProcessor.java
index 40c6e671360..7192f472c14 100644
---
a/extensions/debezium-mysql/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/mysql/deployment/DebeziumMysqlProcessor.java
+++
b/extensions/debezium-mysql/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/mysql/deployment/DebeziumMysqlProcessor.java
@@ -16,15 +16,8 @@
*/
package org.apache.camel.quarkus.component.debezium.mysql.deployment;
-import io.debezium.connector.mysql.MySqlConnector;
-import io.debezium.connector.mysql.MySqlConnectorTask;
-import io.debezium.connector.mysql.MySqlSourceInfoStructMaker;
-import io.debezium.connector.mysql.snapshot.query.SelectAllSnapshotQuery;
-import io.quarkus.deployment.annotations.BuildProducer;
import io.quarkus.deployment.annotations.BuildStep;
import io.quarkus.deployment.builditem.FeatureBuildItem;
-import io.quarkus.deployment.builditem.IndexDependencyBuildItem;
-import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
class DebeziumMysqlProcessor {
@@ -34,20 +27,4 @@ class DebeziumMysqlProcessor {
FeatureBuildItem feature() {
return new FeatureBuildItem(FEATURE);
}
-
- @BuildStep
- void reflectiveClasses(BuildProducer<ReflectiveClassBuildItem>
reflectiveClasses) {
-
- reflectiveClasses.produce(ReflectiveClassBuildItem.builder(
- MySqlConnector.class,
- MySqlConnectorTask.class,
- MySqlSourceInfoStructMaker.class,
- SelectAllSnapshotQuery.class).build());
- }
-
- @BuildStep
- void addDependencies(BuildProducer<IndexDependencyBuildItem>
indexDependency) {
- indexDependency.produce(new IndexDependencyBuildItem("io.debezium",
"debezium-connector-mysql"));
- indexDependency.produce(new IndexDependencyBuildItem("io.debezium",
"debezium-connector-binlog"));
- }
}
diff --git
a/extensions/debezium-oracle/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/oracle/deployment/DebeziumOracleProcessor.java
b/extensions/debezium-oracle/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/oracle/deployment/DebeziumOracleProcessor.java
index bab4e3ae75c..83bf04f7de4 100644
---
a/extensions/debezium-oracle/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/oracle/deployment/DebeziumOracleProcessor.java
+++
b/extensions/debezium-oracle/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/oracle/deployment/DebeziumOracleProcessor.java
@@ -16,16 +16,10 @@
*/
package org.apache.camel.quarkus.component.debezium.oracle.deployment;
-import io.debezium.connector.oracle.OracleConnector;
-import io.debezium.connector.oracle.OracleConnectorTask;
-import io.debezium.connector.oracle.OracleSourceInfoStructMaker;
import io.debezium.connector.oracle.logminer.buffered.BufferedLogMinerAdapter;
-import io.debezium.connector.oracle.snapshot.query.SelectAllSnapshotQuery;
-import io.debezium.storage.kafka.history.KafkaSchemaHistory;
import io.quarkus.deployment.annotations.BuildProducer;
import io.quarkus.deployment.annotations.BuildStep;
import io.quarkus.deployment.builditem.FeatureBuildItem;
-import io.quarkus.deployment.builditem.IndexDependencyBuildItem;
import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
import
io.quarkus.deployment.builditem.nativeimage.RuntimeInitializedClassBuildItem;
import oracle.jdbc.driver.OracleDriver;
@@ -41,20 +35,9 @@ class DebeziumOracleProcessor {
@BuildStep
void reflectiveClasses(BuildProducer<ReflectiveClassBuildItem>
reflectiveClasses) {
-
reflectiveClasses.produce(ReflectiveClassBuildItem.builder(
OracleDriver.class,
- OracleConnector.class,
- OracleConnectorTask.class,
- OracleSourceInfoStructMaker.class,
- SelectAllSnapshotQuery.class,
- BufferedLogMinerAdapter.class,
- KafkaSchemaHistory.class).build());
- }
-
- @BuildStep
- void addDependencies(BuildProducer<IndexDependencyBuildItem>
indexDependency) {
- indexDependency.produce(new IndexDependencyBuildItem("io.debezium",
"debezium-connector-oracle"));
+ BufferedLogMinerAdapter.class).build());
}
@BuildStep
diff --git
a/extensions/debezium-postgres/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/postgres/deployment/DebeziumPostgresProcessor.java
b/extensions/debezium-postgres/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/postgres/deployment/DebeziumPostgresProcessor.java
index d92f552db75..74691435603 100644
---
a/extensions/debezium-postgres/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/postgres/deployment/DebeziumPostgresProcessor.java
+++
b/extensions/debezium-postgres/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/postgres/deployment/DebeziumPostgresProcessor.java
@@ -16,11 +16,8 @@
*/
package org.apache.camel.quarkus.component.debezium.postgres.deployment;
-import io.quarkus.deployment.annotations.BuildProducer;
import io.quarkus.deployment.annotations.BuildStep;
import io.quarkus.deployment.builditem.FeatureBuildItem;
-import io.quarkus.deployment.builditem.IndexDependencyBuildItem;
-import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
import
io.quarkus.deployment.builditem.nativeimage.RuntimeInitializedClassBuildItem;
class DebeziumPostgresProcessor {
@@ -32,25 +29,8 @@ class DebeziumPostgresProcessor {
return new FeatureBuildItem(FEATURE);
}
- @BuildStep
- void reflectiveClasses(BuildProducer<ReflectiveClassBuildItem>
reflectiveClasses) {
- reflectiveClasses.produce(ReflectiveClassBuildItem.builder(
- "io.debezium.connector.postgresql.PostgresConnector",
- "io.debezium.connector.postgresql.PostgresConnectorTask",
-
"io.debezium.connector.postgresql.PostgresSourceInfoStructMaker",
-
"io.debezium.connector.postgresql.snapshot.lock.NoSnapshotLock",
-
"io.debezium.connector.postgresql.snapshot.lock.SharedSnapshotLock",
-
"io.debezium.connector.postgresql.snapshot.query.SelectAllSnapshotQuery")
- .build());
- }
-
@BuildStep
RuntimeInitializedClassBuildItem runtimeInitializedClasses() {
return new
RuntimeInitializedClassBuildItem("com.google.protobuf.JavaFeaturesProto");
}
-
- @BuildStep
- void addDependencies(BuildProducer<IndexDependencyBuildItem>
indexDependency) {
- indexDependency.produce(new IndexDependencyBuildItem("io.debezium",
"debezium-connector-postgres"));
- }
}
diff --git
a/extensions/debezium-sqlserver/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/sqlserver/deployment/DebeziumSqlserverProcessor.java
b/extensions/debezium-sqlserver/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/sqlserver/deployment/DebeziumSqlserverProcessor.java
index f06a9c269d7..2567d1f07ac 100644
---
a/extensions/debezium-sqlserver/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/sqlserver/deployment/DebeziumSqlserverProcessor.java
+++
b/extensions/debezium-sqlserver/deployment/src/main/java/org/apache/camel/quarkus/component/debezium/sqlserver/deployment/DebeziumSqlserverProcessor.java
@@ -16,11 +16,8 @@
*/
package org.apache.camel.quarkus.component.debezium.sqlserver.deployment;
-import io.quarkus.deployment.annotations.BuildProducer;
import io.quarkus.deployment.annotations.BuildStep;
import io.quarkus.deployment.builditem.FeatureBuildItem;
-import io.quarkus.deployment.builditem.IndexDependencyBuildItem;
-import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
class DebeziumSqlserverProcessor {
@@ -30,28 +27,4 @@ class DebeziumSqlserverProcessor {
FeatureBuildItem feature() {
return new FeatureBuildItem(FEATURE);
}
-
- @BuildStep
- ReflectiveClassBuildItem reflectiveClasses() {
- return ReflectiveClassBuildItem.builder(new String[] {
"io.debezium.connector.sqlserver.SqlServerConnector",
- "io.debezium.connector.sqlserver.SqlServerConnectorTask"
}).build();
- }
-
- @BuildStep
- void reflectiveClasses(BuildProducer<ReflectiveClassBuildItem>
reflectiveClasses) {
-
- reflectiveClasses.produce(ReflectiveClassBuildItem.builder(
- "io.debezium.connector.sqlserver.SqlServerConnector",
- "io.debezium.connector.sqlserver.SqlServerConnectorTask",
-
"io.debezium.connector.sqlserver.SqlServerSourceInfoStructMaker",
- "io.debezium.connector.sqlserver.snapshot.lock.NoSnapshotLock",
-
"io.debezium.connector.sqlserver.snapshot.lock.ExclusiveSnapshotLock",
-
"io.debezium.connector.sqlserver.snapshot.query.SelectAllSnapshotQuery")
- .methods().fields().build());
- }
-
- @BuildStep
- void addDependencies(BuildProducer<IndexDependencyBuildItem>
indexDependency) {
- indexDependency.produce(new IndexDependencyBuildItem("io.debezium",
"debezium-connector-sqlserver"));
- }
}
