This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new d617b31304 [Feature][Connector-V2] Foundation for per-connector
Debezium version management (staged) (#10799)
d617b31304 is described below
commit d617b31304c550e86db6d00e4e2bc4c07af1f1fc
Author: Vincenzo Santonastaso <[email protected]>
AuthorDate: Tue Jul 7 15:42:05 2026 +0200
[Feature][Connector-V2] Foundation for per-connector Debezium version
management (staged) (#10799)
Signed-off-by: vsantonastaso <[email protected]>
---
.../cdc/base/debezium/DebeziumAdapter.java | 45 +++++++++
.../cdc/base/debezium/DebeziumAdapterFactory.java | 109 +++++++++++++++++++++
.../base/debezium/DebeziumAdapterFactoryTest.java | 60 ++++++++++++
.../debezium/DuplicateTestDebeziumAdapter.java | 38 +++++++
.../cdc/base/debezium/TestDebeziumAdapter.java | 37 +++++++
...el.connectors.cdc.base.debezium.DebeziumAdapter | 17 ++++
.../connector-cdc/connector-cdc-mongodb/pom.xml | 1 +
.../connector-cdc/connector-cdc-mysql/pom.xml | 4 +
.../connector-cdc/connector-cdc-oracle/pom.xml | 1 +
.../connector-cdc/connector-cdc-postgres/pom.xml | 4 +
.../connector-cdc/connector-cdc-sqlserver/pom.xml | 4 +
11 files changed, 320 insertions(+)
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/debezium/DebeziumAdapter.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/debezium/DebeziumAdapter.java
new file mode 100644
index 0000000000..3a2f3e2f24
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/debezium/DebeziumAdapter.java
@@ -0,0 +1,45 @@
+/*
+ * 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.seatunnel.connectors.cdc.base.debezium;
+
+import org.apache.seatunnel.api.annotation.Experimental;
+
+/**
+ * SPI contract for connector-specific Debezium version adapters.
+ *
+ * <p>Implementations are loaded via {@link java.util.ServiceLoader} from each
connector module's
+ * {@code META-INF/services} registration. This interface defines the target
shape for per-connector
+ * Debezium version management; concrete wiring (schema-history isolation,
classloader separation)
+ * will be added in follow-up PRs once this SPI is established.
+ *
+ * <p>This interface is {@link Experimental} and may change without notice
until the full runtime
+ * wiring lands in a follow-up PR.
+ */
+@Experimental
+public interface DebeziumAdapter {
+
+ /** Returns the Debezium version string this adapter targets (e.g., {@code
"1.9.8.Final"}). */
+ String getDebeziumVersion();
+
+ /**
+ * Returns {@code true} if this adapter handles the given Debezium
connector fully-qualified
+ * class name (e.g., {@code
"io.debezium.connector.mysql.MySqlConnector"}). This matches the
+ * value of the {@code connector.class} Debezium property.
+ */
+ boolean supports(String connectorClassName);
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/debezium/DebeziumAdapterFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/debezium/DebeziumAdapterFactory.java
new file mode 100644
index 0000000000..be08753d4a
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/debezium/DebeziumAdapterFactory.java
@@ -0,0 +1,109 @@
+/*
+ * 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.seatunnel.connectors.cdc.base.debezium;
+
+import org.apache.seatunnel.api.annotation.Experimental;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ServiceLoader;
+import java.util.stream.Collectors;
+
+/**
+ * Factory for loading connector-specific {@link DebeziumAdapter} instances
via {@link
+ * ServiceLoader}.
+ */
+@Experimental
+public class DebeziumAdapterFactory {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(DebeziumAdapterFactory.class);
+
+ private DebeziumAdapterFactory() {}
+
+ /**
+ * Returns the unique {@link DebeziumAdapter} whose {@link
DebeziumAdapter#supports(String)}
+ * method returns {@code true} for the given Debezium connector
fully-qualified class name
+ * (e.g., {@code "io.debezium.connector.mysql.MySqlConnector"}).
+ *
+ * <p>The selection rule is deterministic:
+ *
+ * <ul>
+ * <li>No match → fails with {@link IllegalStateException}
+ * <li>Exactly one match → returns it
+ * <li>More than one match → fails with {@link
IllegalStateException} listing the matched
+ * providers
+ * </ul>
+ *
+ * @param connectorClassName the fully-qualified Debezium connector class
name, matching the
+ * {@code connector.class} Debezium property
+ * @param classLoader the class loader used to discover {@link
DebeziumAdapter} registrations
+ * from {@code META-INF/services}
+ * @throws IllegalStateException if no matching adapter is found, or if
more than one adapter
+ * matches
+ */
+ public static DebeziumAdapter getAdapter(String connectorClassName,
ClassLoader classLoader) {
+ LOG.info("Loading DebeziumAdapter for connector class: {}",
connectorClassName);
+ ServiceLoader<DebeziumAdapter> loader =
+ ServiceLoader.load(DebeziumAdapter.class, classLoader);
+
+ List<DebeziumAdapter> matches = new ArrayList<>();
+ for (DebeziumAdapter adapter : loader) {
+ if (adapter.supports(connectorClassName)) {
+ matches.add(adapter);
+ }
+ }
+
+ if (matches.isEmpty()) {
+ throw new IllegalStateException(
+ "No DebeziumAdapter found for connector class: "
+ + connectorClassName
+ + ". Ensure a META-INF/services/"
+ + DebeziumAdapter.class.getName()
+ + " registration is present in the connector
module.");
+ }
+
+ if (matches.size() > 1) {
+ String providers =
+ matches.stream()
+ .map(
+ a ->
+ a.getClass().getName()
+ + " (Debezium "
+ + a.getDebeziumVersion()
+ + ")")
+ .collect(Collectors.joining(", "));
+ throw new IllegalStateException(
+ "Multiple DebeziumAdapters matched connector class "
+ + connectorClassName
+ + ": ["
+ + providers
+ + "]. Each connector class must have exactly one
adapter.");
+ }
+
+ DebeziumAdapter adapter = matches.get(0);
+ LOG.info(
+ "Found DebeziumAdapter for {}: {} targeting Debezium {}",
+ connectorClassName,
+ adapter.getClass().getName(),
+ adapter.getDebeziumVersion());
+ return adapter;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/debezium/DebeziumAdapterFactoryTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/debezium/DebeziumAdapterFactoryTest.java
new file mode 100644
index 0000000000..5105fa81d5
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/debezium/DebeziumAdapterFactoryTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.seatunnel.connectors.cdc.base.debezium;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class DebeziumAdapterFactoryTest {
+
+ @Test
+ void getAdapter_returnsAdapter_whenExactlyOneMatches() {
+ ClassLoader cl = Thread.currentThread().getContextClassLoader();
+ DebeziumAdapter adapter =
+
DebeziumAdapterFactory.getAdapter(TestDebeziumAdapter.TEST_CONNECTOR_CLASS, cl);
+
+ Assertions.assertInstanceOf(TestDebeziumAdapter.class, adapter);
+ Assertions.assertEquals(
+ TestDebeziumAdapter.TEST_DEBEZIUM_VERSION,
adapter.getDebeziumVersion());
+ }
+
+ @Test
+ void getAdapter_throwsIllegalStateException_whenNoAdapterMatches() {
+ ClassLoader cl = Thread.currentThread().getContextClassLoader();
+ IllegalStateException ex =
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ DebeziumAdapterFactory.getAdapter(
+
"io.debezium.connector.unknown.UnknownConnector", cl));
+ Assertions.assertTrue(ex.getMessage().contains("No DebeziumAdapter
found"));
+ }
+
+ @Test
+ void getAdapter_throwsIllegalStateException_whenMultipleAdaptersMatch() {
+ ClassLoader cl = Thread.currentThread().getContextClassLoader();
+ IllegalStateException ex =
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ DebeziumAdapterFactory.getAdapter(
+
DuplicateTestDebeziumAdapter.DUPLICATE_CONNECTOR_CLASS,
+ cl));
+ Assertions.assertTrue(ex.getMessage().contains("Multiple
DebeziumAdapters matched"));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/debezium/DuplicateTestDebeziumAdapter.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/debezium/DuplicateTestDebeziumAdapter.java
new file mode 100644
index 0000000000..83cd563832
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/debezium/DuplicateTestDebeziumAdapter.java
@@ -0,0 +1,38 @@
+/*
+ * 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.seatunnel.connectors.cdc.base.debezium;
+
+/**
+ * Second test-only {@link DebeziumAdapter} that claims the same connector
class as {@link
+ * TestDebeziumAdapter}, plus a shared duplicate class. Used to verify that
{@link
+ * DebeziumAdapterFactory} rejects ambiguous matches.
+ */
+public class DuplicateTestDebeziumAdapter implements DebeziumAdapter {
+
+ static final String DUPLICATE_CONNECTOR_CLASS =
"io.debezium.connector.duplicate.DupConnector";
+
+ @Override
+ public String getDebeziumVersion() {
+ return "2.0.0.Final";
+ }
+
+ @Override
+ public boolean supports(String connectorClassName) {
+ return DUPLICATE_CONNECTOR_CLASS.equals(connectorClassName);
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/debezium/TestDebeziumAdapter.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/debezium/TestDebeziumAdapter.java
new file mode 100644
index 0000000000..d6a45a043c
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/debezium/TestDebeziumAdapter.java
@@ -0,0 +1,37 @@
+/*
+ * 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.seatunnel.connectors.cdc.base.debezium;
+
+/** Test-only {@link DebeziumAdapter} implementation registered via
META-INF/services. */
+public class TestDebeziumAdapter implements DebeziumAdapter {
+
+ static final String TEST_CONNECTOR_CLASS =
"io.debezium.connector.test.TestConnector";
+ static final String TEST_DEBEZIUM_VERSION = "1.9.8.Final";
+
+ @Override
+ public String getDebeziumVersion() {
+ return TEST_DEBEZIUM_VERSION;
+ }
+
+ @Override
+ public boolean supports(String connectorClassName) {
+ return TEST_CONNECTOR_CLASS.equals(connectorClassName)
+ ||
DuplicateTestDebeziumAdapter.DUPLICATE_CONNECTOR_CLASS.equals(
+ connectorClassName);
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/resources/META-INF/services/org.apache.seatunnel.connectors.cdc.base.debezium.DebeziumAdapter
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/resources/META-INF/services/org.apache.seatunnel.connectors.cdc.base.debezium.DebeziumAdapter
new file mode 100644
index 0000000000..1ddcd28234
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/resources/META-INF/services/org.apache.seatunnel.connectors.cdc.base.debezium.DebeziumAdapter
@@ -0,0 +1,17 @@
+# 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.
+
+org.apache.seatunnel.connectors.cdc.base.debezium.TestDebeziumAdapter
+org.apache.seatunnel.connectors.cdc.base.debezium.DuplicateTestDebeziumAdapter
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/pom.xml
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/pom.xml
index 7734432799..d7a13e164c 100644
--- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/pom.xml
+++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/pom.xml
@@ -29,6 +29,7 @@
<name>SeaTunnel : Connectors V2 : CDC : Mongodb</name>
<properties>
+ <debezium.version>1.9.8.Final</debezium.version>
<mongo.driver.version>4.7.1</mongo.driver.version>
<avro.version>1.11.3</avro.version>
<mongo-kafka-connect.version>1.10.1</mongo-kafka-connect.version>
diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/pom.xml
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/pom.xml
index 701d90f53c..ab9e8a7156 100644
--- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/pom.xml
+++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/pom.xml
@@ -28,6 +28,10 @@
<artifactId>connector-cdc-mysql</artifactId>
<name>SeaTunnel : Connectors V2 : CDC : MySql</name>
+ <properties>
+ <debezium.version>1.9.8.Final</debezium.version>
+ </properties>
+
<dependencyManagement>
<dependencies>
<dependency>
diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/pom.xml
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/pom.xml
index e3a286325d..1ef50a8d83 100644
--- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/pom.xml
+++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/pom.xml
@@ -29,6 +29,7 @@
<name>SeaTunnel : Connectors V2 : CDC : Oracle</name>
<properties>
+ <debezium.version>1.9.8.Final</debezium.version>
<oracle.version>19.18.0.0</oracle.version>
</properties>
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/pom.xml
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/pom.xml
index 61c52fc5bb..3d16cbe8f8 100644
--- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/pom.xml
+++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/pom.xml
@@ -29,6 +29,10 @@
<artifactId>connector-cdc-postgres</artifactId>
<name>SeaTunnel : Connectors V2 : CDC : Postgres</name>
+ <properties>
+ <debezium.version>1.9.8.Final</debezium.version>
+ </properties>
+
<dependencyManagement>
<dependencies>
<dependency>
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/pom.xml
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/pom.xml
index 9865b8c07d..499cef9c36 100644
--- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/pom.xml
+++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/pom.xml
@@ -28,6 +28,10 @@
<artifactId>connector-cdc-sqlserver</artifactId>
<name>SeaTunnel : Connectors V2 : CDC : SqlServer</name>
+ <properties>
+ <debezium.version>1.9.8.Final</debezium.version>
+ </properties>
+
<dependencyManagement>
<dependencies>
<dependency>
