This is an automated email from the ASF dual-hosted git repository. github-bot pushed a commit to branch quarkus-main in repository https://gitbox.apache.org/repos/asf/camel-quarkus.git
commit 242e702bd24829428335eed266c8dd30978a0c27 Author: James Netherton <[email protected]> AuthorDate: Tue Apr 14 15:44:34 2026 +0100 Work around Debezium kafka-clients 4.2.0 incompatibility due to #8530 Co-Authored-By: Claude Sonnet 4.5 <[email protected]> --- .../deployment/DebeziumSupportProcessor.java | 184 +++++++++++++++++++++ 1 file changed, 184 insertions(+) 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..acc8fe55d0 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,6 +16,8 @@ */ package org.apache.camel.quarkus.support.debezium.deployment; +import java.util.function.BooleanSupplier; + import io.debezium.connector.common.BaseSourceTask; import io.debezium.embedded.async.ConvertingAsyncEngineBuilderFactory; import io.debezium.engine.DebeziumEngine; @@ -43,19 +45,27 @@ import io.debezium.snapshot.spi.SnapshotLock; import io.debezium.storage.file.history.FileSchemaHistory; import io.quarkus.deployment.annotations.BuildProducer; import io.quarkus.deployment.annotations.BuildStep; +import io.quarkus.deployment.builditem.BytecodeTransformerBuildItem; 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.RuntimeInitializedClassBuildItem; +import io.quarkus.gizmo.Gizmo; import org.apache.kafka.common.security.authenticator.SaslClientAuthenticator; import org.apache.kafka.connect.json.JsonConverter; import org.apache.kafka.connect.source.SourceTask; import org.jboss.jandex.DotName; import org.jboss.jandex.IndexView; +import org.jboss.logging.Logger; +import org.objectweb.asm.ClassVisitor; +import org.objectweb.asm.MethodVisitor; +import org.objectweb.asm.Opcodes; public class DebeziumSupportProcessor { + private static final Logger LOG = Logger.getLogger(DebeziumSupportProcessor.class); + @BuildStep void addDependencies(BuildProducer<IndexDependencyBuildItem> indexDependency) { indexDependency.produce(new IndexDependencyBuildItem("org.apache.kafka", "connect-json")); @@ -148,4 +158,178 @@ public class DebeziumSupportProcessor { .produce(new NativeImageResourceBuildItem("META-INF/services/org.apache.kafka.connect.source.SourceConnector")); } + // TODO: Remove this - https://github.com/apache/camel-quarkus/issues/8530 + @BuildStep(onlyIf = DebeziumKafkaClients42SupportIsAbsent.class) + BytecodeTransformerBuildItem patchWorkerConfig() { + // Patch WorkerConfig.baseConfigDef() to work with kafka-clients 4.2.0 + // Changes bootstrap.servers default value from NO_DEFAULT_VALUE to "localhost:9092" as per kafka-clients 4.1.x + // This allows Debezium embedded engine to work without requiring bootstrap.servers configuration + return new BytecodeTransformerBuildItem.Builder() + .setClassToTransform("org.apache.kafka.connect.runtime.WorkerConfig") + .setCacheable(true) + .setVisitorFunction((className, classVisitor) -> new WorkerConfigClassVisitor(classVisitor)) + .build(); + } + + // TODO: Remove this - https://github.com/apache/camel-quarkus/issues/8530 + @BuildStep(onlyIf = DebeziumKafkaClients42SupportIsAbsent.class) + BytecodeTransformerBuildItem patchConfigInfos() { + // Patch ConfigInfos to add values() method as duplicate of configs() + // This provides backward compatibility for Debezium with kafka-clients 4.2.0 + return new BytecodeTransformerBuildItem.Builder() + .setClassToTransform("org.apache.kafka.connect.runtime.rest.entities.ConfigInfos") + .setCacheable(true) + .setVisitorFunction((className, classVisitor) -> new ConfigInfosClassVisitor(classVisitor)) + .build(); + } + + // TODO: Remove this - https://github.com/apache/camel-quarkus/issues/8530 + static final class DebeziumKafkaClients42SupportIsAbsent implements BooleanSupplier { + @Override + public boolean getAsBoolean() { + try { + // Based on https://github.com/debezium/debezium/pull/7105 + Thread.currentThread().getContextClassLoader().loadClass("io.debezium.util.Reflections"); + return false; + } catch (ClassNotFoundException e) { + return true; + } + } + } + + /** + * Modifies WorkerConfig.baseConfigDef() to change the default value for bootstrap.servers + * from NO_DEFAULT_VALUE to "localhost:9092" for compatibility with kafka-clients 4.2.0. + * + * This allows Debezium embedded engine to work without requiring bootstrap.servers configuration. + */ + static class WorkerConfigClassVisitor extends ClassVisitor { + + protected WorkerConfigClassVisitor(ClassVisitor classVisitor) { + super(Gizmo.ASM_API_VERSION, classVisitor); + } + + @Override + public MethodVisitor visitMethod(int access, String name, String descriptor, String signature, + String[] exceptions) { + MethodVisitor original = super.visitMethod(access, name, descriptor, signature, exceptions); + if ("baseConfigDef".equals(name)) { + LOG.debug("Transforming WorkerConfig.baseConfigDef() for Kafka 4.2.0 compatibility"); + return new BootstrapServersDefaultValueVisitor(original); + } + return original; + } + } + + /** + * Adds a values() method to ConfigInfos that duplicates the configs() method. + * This provides backward compatibility with older Debezium versions. + */ + static class ConfigInfosClassVisitor extends ClassVisitor { + + private String configsFieldDescriptor = null; + private String configsMethodSignature = null; + + protected ConfigInfosClassVisitor(ClassVisitor classVisitor) { + super(Gizmo.ASM_API_VERSION, classVisitor); + } + + @Override + public org.objectweb.asm.FieldVisitor visitField(int access, String name, String descriptor, String signature, + Object value) { + // Track the configs field descriptor + if ("configs".equals(name)) { + configsFieldDescriptor = descriptor; + } + return super.visitField(access, name, descriptor, signature, value); + } + + @Override + public MethodVisitor visitMethod(int access, String name, String descriptor, String signature, + String[] exceptions) { + // Track the signature of configs() method + if ("configs".equals(name) && "()Ljava/util/List;".equals(descriptor)) { + configsMethodSignature = signature; + } + return super.visitMethod(access, name, descriptor, signature, exceptions); + } + + @Override + public void visitEnd() { + // Add values() method that duplicates configs() + LOG.debug("Adding values() method to ConfigInfos as duplicate of configs()"); + + MethodVisitor mv = cv.visitMethod( + Opcodes.ACC_PUBLIC, + "values", + "()Ljava/util/List;", + configsMethodSignature, // Same generic signature as configs() + null); + + if (mv != null) { + mv.visitCode(); + // Method body: return this.configs; + mv.visitVarInsn(Opcodes.ALOAD, 0); // Load 'this' + mv.visitFieldInsn(Opcodes.GETFIELD, + "org/apache/kafka/connect/runtime/rest/entities/ConfigInfos", + "configs", + configsFieldDescriptor != null ? configsFieldDescriptor : "Ljava/util/List;"); + mv.visitInsn(Opcodes.ARETURN); // Return the field value + mv.visitMaxs(1, 1); // Max stack=1, max locals=1 + mv.visitEnd(); + } + + super.visitEnd(); + } + } + + /** + * Method visitor that replaces the default value for bootstrap.servers + * from NO_DEFAULT_VALUE to "localhost:9092". + */ + static class BootstrapServersDefaultValueVisitor extends MethodVisitor { + + private boolean seenBootstrapServersKey = false; + + protected BootstrapServersDefaultValueVisitor(MethodVisitor methodVisitor) { + super(Gizmo.ASM_API_VERSION, methodVisitor); + } + + @Override + public void visitLdcInsn(Object value) { + // Track when we see the "bootstrap.servers" string constant + if ("bootstrap.servers".equals(value)) { + seenBootstrapServersKey = true; + LOG.debug("Found bootstrap.servers config key"); + } + super.visitLdcInsn(value); + } + + @Override + public void visitFieldInsn(int opcode, String owner, String name, String descriptor) { + // Replace NO_DEFAULT_VALUE with "localhost:9092" for bootstrap.servers + if (seenBootstrapServersKey + && opcode == Opcodes.GETSTATIC + && "org/apache/kafka/common/config/ConfigDef".equals(owner) + && "NO_DEFAULT_VALUE".equals(name)) { + + LOG.debugf("Replacing NO_DEFAULT_VALUE with \"localhost:9092\" for bootstrap.servers"); + // Instead of loading NO_DEFAULT_VALUE, load the string "localhost:9092" + super.visitLdcInsn("localhost:9092"); + seenBootstrapServersKey = false; // Reset flag + } else { + super.visitFieldInsn(opcode, owner, name, descriptor); + } + } + + @Override + public void visitMethodInsn(int opcode, String owner, String name, String descriptor, boolean isInterface) { + // Reset flag when we see a define() call (end of current config definition) + if ("define".equals(name) && "org/apache/kafka/common/config/ConfigDef".equals(owner)) { + seenBootstrapServersKey = false; + } + super.visitMethodInsn(opcode, owner, name, descriptor, isInterface); + } + } + }
