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 523d69dbb1 [Feature][Connector-V2] Add OceanBase CDC source connector
(#11060)
523d69dbb1 is described below
commit 523d69dbb1e324ab58ead6178b1b2a3a7c327aa4
Author: David Zollo <[email protected]>
AuthorDate: Tue Sep 22 09:40:56 2026 -0400
[Feature][Connector-V2] Add OceanBase CDC source connector (#11060)
Co-authored-by: Daniel <[email protected]>
Co-authored-by: Daniel <[email protected]>
---
config/plugin_config | 1 +
.../changelog/connector-cdc-oceanbase.md | 6 +
docs/en/connectors/source/OceanBase-CDC.md | 119 +++++++++
.../changelog/connector-cdc-oceanbase.md | 6 +
docs/zh/connectors/source/OceanBase-CDC.md | 117 +++++++++
plugin-mapping.properties | 1 +
.../source/MySqlIncrementalSourceFactory.java | 108 ++++----
.../connector-cdc/connector-cdc-oceanbase/pom.xml | 46 ++++
.../source/OceanBaseIncrementalSource.java | 59 +++++
.../source/OceanBaseIncrementalSourceFactory.java | 123 +++++++++
.../OceanBaseIncrementalSourceFactoryTest.java | 109 ++++++++
seatunnel-connectors-v2/connector-cdc/pom.xml | 1 +
seatunnel-dist/pom.xml | 6 +
.../connector-cdc-oceanbase-e2e/pom.xml | 69 +++++
.../cdc/oceanbase/MySqlCompatibleContainer.java | 287 ++++++++++++++++++++
.../cdc/oceanbase/OceanBaseCDCCompatibilityIT.java | 288 +++++++++++++++++++++
.../src/test/resources/docker/server-gtids/my.cnf | 31 +++
.../src/test/resources/docker/setup.sql | 24 ++
.../oceanbasecdc_to_mysql_compatible.conf | 49 ++++
seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml | 1 +
20 files changed, 1406 insertions(+), 45 deletions(-)
diff --git a/config/plugin_config b/config/plugin_config
index 651cb4033b..2636f8f827 100644
--- a/config/plugin_config
+++ b/config/plugin_config
@@ -29,6 +29,7 @@ connector-cassandra
connector-salesforce
connector-cdc-db2
connector-cdc-mysql
+connector-cdc-oceanbase
connector-cdc-mongodb
connector-cdc-sqlserver
connector-cdc-postgres
diff --git a/docs/en/connectors/changelog/connector-cdc-oceanbase.md
b/docs/en/connectors/changelog/connector-cdc-oceanbase.md
new file mode 100644
index 0000000000..1cf41e674b
--- /dev/null
+++ b/docs/en/connectors/changelog/connector-cdc-oceanbase.md
@@ -0,0 +1,6 @@
+<details><summary> Change Log </summary>
+
+| Change | Commit | Version |
+| --- | --- | --- |
+
+</details>
diff --git a/docs/en/connectors/source/OceanBase-CDC.md
b/docs/en/connectors/source/OceanBase-CDC.md
new file mode 100644
index 0000000000..da02447149
--- /dev/null
+++ b/docs/en/connectors/source/OceanBase-CDC.md
@@ -0,0 +1,119 @@
+import ChangeLog from '../changelog/connector-cdc-oceanbase.md';
+
+# OceanBase CDC
+
+> OceanBase CDC source connector
+
+## Support Those Engines
+
+> SeaTunnel Zeta<br/>
+> Flink <br/>
+
+## Key features
+
+- [ ] [batch](../../introduction/concepts/connector-v2-features.md)
+- [x] [stream](../../introduction/concepts/connector-v2-features.md)
+- [x] [exactly-once](../../introduction/concepts/connector-v2-features.md)
+- [ ] [column projection](../../introduction/concepts/connector-v2-features.md)
+- [x] [parallelism](../../introduction/concepts/connector-v2-features.md)
+- [x] [support user-defined
split](../../introduction/concepts/connector-v2-features.md)
+
+## Description
+
+The first delivery of the OceanBase CDC connector targets the stable OceanBase
MySQL compatible
+path backed by OceanBase Binlog Service.
+
+To keep the implementation narrow and restart-safe, `OceanBase-CDC` reuses
SeaTunnel's
+`MySQL-CDC` incremental runtime for:
+
+- snapshot + incremental capture
+- checkpoint / restore handling
+- multi-table CDC row semantics
+- schema change propagation supported by `MySQL-CDC`
+
+OceanBase Oracle compatible mode is not supported in this first delivery.
+
+## Supported DataSource Info
+
+| Datasource | Supported versions | Driver | Url | Notes |
+| --- | --- | --- | --- | --- |
+| OceanBase CE / OceanBase EE (MySQL compatible mode) | OceanBase deployments
that expose a MySQL-compatible snapshot endpoint and OceanBase Binlog Service |
`com.mysql.cj.jdbc.Driver` | `jdbc:mysql://localhost:2881/test` | Requires
OceanBase Binlog Service |
+
+## Using Dependency
+
+### Install Jdbc Driver
+
+#### For Flink Engine
+
+> 1. You need to ensure that the [MySQL JDBC
driver](https://mvnrepository.com/artifact/mysql/mysql-connector-java) has been
placed in `${SEATUNNEL_HOME}/plugins/`.
+
+#### For SeaTunnel Zeta Engine
+
+> 1. You need to ensure that the [MySQL JDBC
driver](https://mvnrepository.com/artifact/mysql/mysql-connector-java) has been
placed in `${SEATUNNEL_HOME}/lib/`.
+
+## OceanBase Preparation
+
+Before using `OceanBase-CDC`, make sure the monitored tenant satisfies these
requirements:
+
+1. OceanBase runs in MySQL compatible mode for the captured tables.
+2. OceanBase Binlog Service is deployed and enabled for incremental
subscription.
+3. The JDBC `url` points to a MySQL-compatible endpoint that SeaTunnel can use
for snapshot reads.
+4. The configured account can read the captured tables and subscribe to
incremental changes.
+
+## Source Options
+
+`OceanBase-CDC` reuses the `MySQL-CDC` option contract for its CDC behavior.
+
+Please refer to [MySQL CDC Source Options](./MySQL-CDC.md#source-options) for
the complete option
+list.
+
+### OceanBase-specific constraints
+
+- Use a MySQL-compatible JDBC URL such as `jdbc:mysql://host:2881/database`.
+- Use the MySQL JDBC driver `com.mysql.cj.jdbc.Driver`.
+- `compatible_mode` is optional and is fixed to `mysql` when omitted. If
configured, it must be
+ `mysql`; Oracle-compatible mode is not supported by this connector.
+- The first delivery supports explicitly configured tables only through
`table-names`,
+ `table-pattern`, and `table-names-config`.
+- Startup modes, checkpoint / restore semantics, and schema evolution behavior
are the same as
+ `MySQL-CDC`.
+
+## Task Example
+
+### Simple
+
+> Support multi-table reading from OceanBase MySQL compatible mode
+
+```
+env {
+ execution.parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ OceanBase-CDC {
+ plugin_output = "oceanbase_cdc"
+ username = "root"
+ password = "123456"
+ url = "jdbc:mysql://127.0.0.1:2881/inventory"
+ database-names = ["inventory"]
+ table-names = ["inventory.orders", "inventory.customers"]
+ server-time-zone = "Asia/Shanghai"
+ startup.mode = "initial"
+ exactly_once = true
+ }
+}
+
+transform {
+
+}
+
+sink {
+ Console {}
+}
+```
+
+## Changelog
+
+<ChangeLog />
diff --git a/docs/zh/connectors/changelog/connector-cdc-oceanbase.md
b/docs/zh/connectors/changelog/connector-cdc-oceanbase.md
new file mode 100644
index 0000000000..527d6aa9c8
--- /dev/null
+++ b/docs/zh/connectors/changelog/connector-cdc-oceanbase.md
@@ -0,0 +1,6 @@
+<details><summary> Change Log </summary>
+
+| 变更 | Commit | 版本 |
+| --- | --- | --- |
+
+</details>
diff --git a/docs/zh/connectors/source/OceanBase-CDC.md
b/docs/zh/connectors/source/OceanBase-CDC.md
new file mode 100644
index 0000000000..70d7fe9a52
--- /dev/null
+++ b/docs/zh/connectors/source/OceanBase-CDC.md
@@ -0,0 +1,117 @@
+import ChangeLog from '../changelog/connector-cdc-oceanbase.md';
+
+# OceanBase CDC
+
+> OceanBase CDC 源连接器
+
+## 支持这些引擎
+
+> SeaTunnel Zeta<br/>
+> Flink <br/>
+
+## 主要功能
+
+- [ ] [批处理](../../introduction/concepts/connector-v2-features.md)
+- [x] [流处理](../../introduction/concepts/connector-v2-features.md)
+- [x] [精确一次](../../introduction/concepts/connector-v2-features.md)
+- [ ] [列投影](../../introduction/concepts/connector-v2-features.md)
+- [x] [并行度](../../introduction/concepts/connector-v2-features.md)
+- [x] [支持用户定义的拆分](../../introduction/concepts/connector-v2-features.md)
+
+## 描述
+
+OceanBase CDC 首个交付版本只覆盖稳定、可验证的 OceanBase MySQL 兼容模式路径，并依赖
+OceanBase Binlog Service 提供增量订阅能力。
+
+为了把实现范围控制在最小且确保重启恢复语义可靠，`OceanBase-CDC` 直接复用 SeaTunnel
+现有 `MySQL-CDC` 的增量运行时，包括：
+
+- 全量快照 + 增量读取
+- checkpoint / restore 处理
+- 多表 CDC 行语义
+- `MySQL-CDC` 已支持的模式演进能力
+
+本次首批交付暂不支持 OceanBase Oracle 兼容模式的增量 CDC。
+
+## 支持的数据源信息
+
+| 数据源 | 支持版本 | 驱动 | URL | 说明 |
+| --- | --- | --- | --- | --- |
+| OceanBase CE / OceanBase EE（MySQL 兼容模式） | 暴露 MySQL 兼容快照读取端点，并开启 OceanBase
Binlog Service 的部署 | `com.mysql.cj.jdbc.Driver` |
`jdbc:mysql://localhost:2881/test` | 依赖 OceanBase Binlog Service |
+
+## 依赖使用
+
+### 安装 JDBC 驱动
+
+#### 对于 Flink 引擎
+
+> 1. 你需要确保 [MySQL JDBC
驱动](https://mvnrepository.com/artifact/mysql/mysql-connector-java) 已经放在
`${SEATUNNEL_HOME}/plugins/` 目录下。
+
+#### 对于 SeaTunnel Zeta 引擎
+
+> 1. 你需要确保 [MySQL JDBC
驱动](https://mvnrepository.com/artifact/mysql/mysql-connector-java) 已经放在
`${SEATUNNEL_HOME}/lib/` 目录下。
+
+## OceanBase 前置准备
+
+在使用 `OceanBase-CDC` 之前，请确保被采集租户满足以下条件：
+
+1. OceanBase 运行在 MySQL 兼容模式。
+2. OceanBase Binlog Service 已部署并开启增量订阅能力。
+3. JDBC `url` 指向 SeaTunnel 可用于快照读取的 MySQL 兼容端点。
+4. 配置账号同时具备快照读取和增量订阅所需权限。
+
+## 源端可选项
+
+`OceanBase-CDC` 在 CDC 行为上复用 `MySQL-CDC` 的配置契约。
+
+完整参数请直接参考 [MySQL CDC 源端参数](./MySQL-CDC.md#配置参数选项)。
+
+### OceanBase 特有约束
+
+- JDBC URL 需要使用 MySQL 兼容写法，例如 `jdbc:mysql://host:2881/database`。
+- JDBC 驱动固定使用 MySQL 驱动 `com.mysql.cj.jdbc.Driver`。
+- `compatible_mode` 为可选项，未配置时固定为 `mysql`。如显式配置，其值必须为
+ `mysql`；该连接器不支持 Oracle 兼容模式。
+- 首批版本只支持显式配置的表，即通过 `table-names`、`table-pattern`、
+ `table-names-config` 指定采集范围。
+- 启动模式、checkpoint / restore 语义、模式演进行为与 `MySQL-CDC` 保持一致。
+
+## 任务示例
+
+### 简单示例
+
+> 读取 OceanBase MySQL 兼容模式下的多张表
+
+```
+env {
+ execution.parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ OceanBase-CDC {
+ plugin_output = "oceanbase_cdc"
+ username = "root"
+ password = "123456"
+ url = "jdbc:mysql://127.0.0.1:2881/inventory"
+ database-names = ["inventory"]
+ table-names = ["inventory.orders", "inventory.customers"]
+ server-time-zone = "Asia/Shanghai"
+ startup.mode = "initial"
+ exactly_once = true
+ }
+}
+
+transform {
+
+}
+
+sink {
+ Console {}
+}
+```
+
+## 变更日志
+
+<ChangeLog />
diff --git a/plugin-mapping.properties b/plugin-mapping.properties
index f77a63baf4..4b41bb59a4 100644
--- a/plugin-mapping.properties
+++ b/plugin-mapping.properties
@@ -139,6 +139,7 @@ seatunnel.sink.Doris = connector-doris
seatunnel.source.Maxcompute = connector-maxcompute
seatunnel.sink.Maxcompute = connector-maxcompute
seatunnel.source.MySQL-CDC = connector-cdc-mysql
+seatunnel.source.OceanBase-CDC = connector-cdc-oceanbase
seatunnel.source.MongoDB-CDC = connector-cdc-mongodb
seatunnel.source.TiDB-CDC = connector-cdc-tidb
seatunnel.source.Vitess-CDC = connector-cdc-vitess
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactory.java
index 9f4a45fc04..b16b0ddfc1 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactory.java
@@ -54,6 +54,16 @@ public class MySqlIncrementalSourceFactory extends
BaseChangeStreamTableSourceFa
@Override
public OptionRule optionRule() {
+ return getOptionRuleBuilder().build();
+ }
+
+ /**
+ * Create the common MySQL CDC source option rule builder.
+ *
+ * <p>Subclasses that reuse the MySQL CDC runtime can add
connector-specific options before
+ * building the rule.
+ */
+ protected OptionRule.Builder getOptionRuleBuilder() {
return JdbcSourceOptions.getBaseRule()
.required(
MySqlIncrementalSourceOptions.USERNAME,
@@ -101,8 +111,7 @@ public class MySqlIncrementalSourceFactory extends
BaseChangeStreamTableSourceFa
.conditional(
MySqlIncrementalSourceOptions.STARTUP_MODE,
StartupMode.TIMESTAMP,
- SourceOptions.STARTUP_TIMESTAMP)
- .build();
+ SourceOptions.STARTUP_TIMESTAMP);
}
@Override
@@ -115,51 +124,60 @@ public class MySqlIncrementalSourceFactory extends
BaseChangeStreamTableSourceFa
TableSource<T, SplitT, StateT> restoreSource(
TableSourceFactoryContext context, List<CatalogTable>
restoreTables) {
return () -> {
- // Load the JDBC driver in to DriverManager
- try {
- Class.forName("com.mysql.cj.jdbc.Driver");
- } catch (Exception e) {
- log.warn("Failed to load JDBC driver com.mysql.cj.jdbc.Driver
", e);
- }
ReadonlyConfig config = context.getOptions();
- List<CatalogTable> catalogTables =
- CatalogTableUtil.getCatalogTables(config,
context.getClassLoader());
- boolean enableSchemaChange =
- context.getOptions()
- .getOptional(SourceOptions.SCHEMA_CHANGES_ENABLED)
- .orElse(
- // TODO remove this after all users used
the new schema change
- // option
- context.getOptions()
-
.getOptional(SourceOptions.DEBEZIUM_PROPERTIES)
- .map(
- e ->
- e.getOrDefault(
-
MySqlSourceConfigFactory
-
.SCHEMA_CHANGE_KEY,
-
SourceOptions
-
.SCHEMA_CHANGES_ENABLED
-
.defaultValue()
-
.toString()))
- .map(Boolean::parseBoolean)
- .orElse(
-
SourceOptions.SCHEMA_CHANGES_ENABLED
- .defaultValue()));
- if (!restoreTables.isEmpty() && enableSchemaChange) {
- catalogTables = mergeTableStruct(catalogTables, restoreTables);
- }
-
- Optional<List<JdbcSourceTableConfig>> tableConfigs =
-
context.getOptions().getOptional(JdbcSourceOptions.TABLE_NAMES_CONFIG);
- if (tableConfigs.isPresent()) {
- catalogTables =
- CatalogTableUtils.mergeCatalogTableConfig(
- catalogTables,
- tableConfigs.get(),
- text -> TablePath.of(text, false));
- }
return (SeaTunnelSource<T, SplitT, StateT>)
- new MySqlIncrementalSource<>(config, catalogTables);
+ new MySqlIncrementalSource<>(
+ config, buildCatalogTables(context, config,
restoreTables));
};
}
+
+ /**
+ * Build source tables from the current configuration and checkpoint state.
+ *
+ * <p>Subclasses that reuse the MySQL CDC runtime can provide a normalized
connector
+ * configuration while keeping catalog discovery and checkpoint merge
behavior aligned.
+ */
+ protected List<CatalogTable> buildCatalogTables(
+ TableSourceFactoryContext context,
+ ReadonlyConfig config,
+ List<CatalogTable> restoreTables) {
+ // Load the JDBC driver in to DriverManager
+ try {
+ Class.forName("com.mysql.cj.jdbc.Driver");
+ } catch (Exception e) {
+ log.warn("Failed to load JDBC driver com.mysql.cj.jdbc.Driver ",
e);
+ }
+ List<CatalogTable> catalogTables =
+ CatalogTableUtil.getCatalogTables(config,
context.getClassLoader());
+ boolean enableSchemaChange =
+ config.getOptional(SourceOptions.SCHEMA_CHANGES_ENABLED)
+ .orElse(
+ // TODO remove this after all users used the
new schema change
+ // option
+
config.getOptional(SourceOptions.DEBEZIUM_PROPERTIES)
+ .map(
+ e ->
+ e.getOrDefault(
+
MySqlSourceConfigFactory
+
.SCHEMA_CHANGE_KEY,
+
SourceOptions.SCHEMA_CHANGES_ENABLED
+
.defaultValue()
+
.toString()))
+ .map(Boolean::parseBoolean)
+ .orElse(
+
SourceOptions.SCHEMA_CHANGES_ENABLED
+ .defaultValue()));
+ if (!restoreTables.isEmpty() && enableSchemaChange) {
+ catalogTables = mergeTableStruct(catalogTables, restoreTables);
+ }
+
+ Optional<List<JdbcSourceTableConfig>> tableConfigs =
+ config.getOptional(JdbcSourceOptions.TABLE_NAMES_CONFIG);
+ if (tableConfigs.isPresent()) {
+ catalogTables =
+ CatalogTableUtils.mergeCatalogTableConfig(
+ catalogTables, tableConfigs.get(), text ->
TablePath.of(text, false));
+ }
+ return catalogTables;
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/pom.xml
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/pom.xml
new file mode 100644
index 0000000000..460f5c07b3
--- /dev/null
+++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/pom.xml
@@ -0,0 +1,46 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-cdc</artifactId>
+ <version>${revision}</version>
+ </parent>
+
+ <artifactId>connector-cdc-oceanbase</artifactId>
+ <name>SeaTunnel : Connectors V2 : CDC : OceanBase</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-cdc-base</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-cdc-mysql</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+</project>
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSource.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSource.java
new file mode 100644
index 0000000000..36a020556d
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSource.java
@@ -0,0 +1,59 @@
+/*
+ * 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.seatunnel.cdc.oceanbase.source;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.MySqlIncrementalSource;
+
+import java.util.List;
+
+/**
+ * OceanBase CDC source for the first delivery path that targets OceanBase
MySQL compatible mode.
+ *
+ * <p>The connector intentionally reuses the MySQL CDC runtime because
OceanBase Binlog Service
+ * exposes a MySQL-compatible change-log interface, which keeps snapshot,
incremental, checkpoint,
+ * and restore semantics aligned with the existing MySQL connector.
+ *
+ * @param <T> emitted record type
+ */
+public class OceanBaseIncrementalSource<T> extends MySqlIncrementalSource<T> {
+
+ /** Stable plugin identifier used by SeaTunnel config and plugin
discovery. */
+ static final String IDENTIFIER = "OceanBase-CDC";
+
+ /**
+ * Create an OceanBase CDC source backed by the MySQL CDC runtime.
+ *
+ * @param options connector options
+ * @param catalogTables tables discovered or restored for the source
+ */
+ public OceanBaseIncrementalSource(ReadonlyConfig options,
List<CatalogTable> catalogTables) {
+ super(options, catalogTables);
+ }
+
+ /**
+ * Expose the OceanBase-specific plugin name while reusing
MySQL-compatible internals.
+ *
+ * @return OceanBase CDC plugin name
+ */
+ @Override
+ public String getPluginName() {
+ return IDENTIFIER;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceFactory.java
new file mode 100644
index 0000000000..3c5599d90e
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceFactory.java
@@ -0,0 +1,123 @@
+/*
+ * 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.seatunnel.cdc.oceanbase.source;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.Conditions;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
+import org.apache.seatunnel.api.source.SourceSplit;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.connector.TableSource;
+import org.apache.seatunnel.api.table.factory.Factory;
+import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.MySqlIncrementalSourceFactory;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions;
+
+import com.google.auto.service.AutoService;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Factory for the OceanBase CDC source.
+ *
+ * <p>The first implementation deliberately wraps the MySQL CDC connector
because that is the stable
+ * and testable path already adopted by Flink CDC for OceanBase Binlog Service.
+ */
+@AutoService(Factory.class)
+public class OceanBaseIncrementalSourceFactory extends
MySqlIncrementalSourceFactory {
+
+ private static final String MYSQL_COMPATIBLE_MODE = "mysql";
+
+ /**
+ * Return the identifier used in SeaTunnel source config.
+ *
+ * @return OceanBase CDC identifier
+ */
+ @Override
+ public String factoryIdentifier() {
+ return OceanBaseIncrementalSource.IDENTIFIER;
+ }
+
+ /**
+ * Return the concrete source class used for OceanBase CDC jobs.
+ *
+ * @return OceanBase CDC source class
+ */
+ @Override
+ public Class<? extends SeaTunnelSource> getSourceClass() {
+ return OceanBaseIncrementalSource.class;
+ }
+
+ /**
+ * Allow the OceanBase catalog selector while restricting this
MySQL-binlog wrapper to MySQL
+ * compatible mode.
+ */
+ @Override
+ public OptionRule optionRule() {
+ return getOptionRuleBuilder()
+ .optional(
+ JdbcCommonOptions.COMPATIBLE_MODE,
+ Conditions.matches(JdbcCommonOptions.COMPATIBLE_MODE,
"(?i)mysql"))
+ .build();
+ }
+
+ /**
+ * Restore the source by reusing MySQL-compatible table discovery and
checkpoint merge logic.
+ *
+ * @param context source factory context
+ * @param restoreTables restored table structures from checkpoint state
+ * @param <T> emitted record type
+ * @param <SplitT> split type
+ * @param <StateT> state type
+ * @return restorable OceanBase CDC table source
+ */
+ @Override
+ public <T, SplitT extends SourceSplit, StateT extends Serializable>
+ TableSource<T, SplitT, StateT> restoreSource(
+ TableSourceFactoryContext context, List<CatalogTable>
restoreTables) {
+ return () -> {
+ ReadonlyConfig config =
mysqlCompatibleConfig(context.getOptions());
+ return (SeaTunnelSource<T, SplitT, StateT>)
+ new OceanBaseIncrementalSource<>(
+ config, buildCatalogTables(context, config,
restoreTables));
+ };
+ }
+
+ /**
+ * Add the catalog selector required by OceanBase catalog discovery when
users omit it.
+ *
+ * <p>Validation permits only MySQL mode because the incremental runtime
uses MySQL binlog
+ * semantics.
+ */
+ ReadonlyConfig mysqlCompatibleConfig(ReadonlyConfig config) {
+ String compatibleMode =
+
config.getOptional(JdbcCommonOptions.COMPATIBLE_MODE).orElse(MYSQL_COMPATIBLE_MODE);
+ if (!MYSQL_COMPATIBLE_MODE.equalsIgnoreCase(compatibleMode)) {
+ throw new IllegalArgumentException(
+ "OceanBase-CDC supports only MySQL compatible mode, but
compatible_mode is "
+ + compatibleMode);
+ }
+ Map<String, Object> options = new HashMap<>(config.getSourceMap());
+ options.put(JdbcCommonOptions.COMPATIBLE_MODE.key(),
MYSQL_COMPATIBLE_MODE);
+ return ReadonlyConfig.fromMap(options);
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceFactoryTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceFactoryTest.java
new file mode 100644
index 0000000000..9878a05024
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceFactoryTest.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.seatunnel.cdc.oceanbase.source;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConfigValidator;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.MySqlIncrementalSourceFactory;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Tests the OceanBase CDC wrapper contract that differentiates it from the
reused MySQL CDC
+ * implementation.
+ */
+public class OceanBaseIncrementalSourceFactoryTest {
+
+ /** Verify the OceanBase wrapper uses a dedicated factory identifier for
plugin discovery. */
+ @Test
+ public void testFactoryIdentifier() {
+ Assertions.assertEquals(
+ "OceanBase-CDC", new
OceanBaseIncrementalSourceFactory().factoryIdentifier());
+ }
+
+ /** Verify the wrapper keeps the MySQL factory inheritance so restore
logic stays aligned. */
+ @Test
+ public void testFactoryInheritance() {
+ Assertions.assertEquals(
+ MySqlIncrementalSourceFactory.class,
+ OceanBaseIncrementalSourceFactory.class.getSuperclass());
+ }
+
+ /** Verify the factory returns the OceanBase wrapper source instead of the
raw MySQL source. */
+ @Test
+ public void testSourceClass() {
+ Assertions.assertEquals(
+ OceanBaseIncrementalSource.class,
+ new OceanBaseIncrementalSourceFactory().getSourceClass());
+ }
+
+ /** Verify restore configuration defaults the OceanBase catalog selector
to MySQL mode. */
+ @Test
+ public void testRestoreConfigurationDefaultsCatalogCompatibleModeToMysql()
{
+ OceanBaseIncrementalSourceFactory factory = new
OceanBaseIncrementalSourceFactory();
+ ReadonlyConfig config = ReadonlyConfig.fromMap(requiredOptions());
+
+ ConfigValidator.of(config).validate(factory.optionRule());
+ ConfigValidator.validateUnknownKeys(config, factory.optionRule(),
"OceanBase-CDC");
+
+ Assertions.assertEquals(
+ "mysql",
+
factory.mysqlCompatibleConfig(config).get(JdbcCommonOptions.COMPATIBLE_MODE));
+ }
+
+ /** Verify static validation accepts the explicit MySQL selector used by
E2E configurations. */
+ @Test
+ public void testOptionRuleAcceptsMysqlCompatibleMode() {
+ OceanBaseIncrementalSourceFactory factory = new
OceanBaseIncrementalSourceFactory();
+ Map<String, Object> options = requiredOptions();
+ options.put(JdbcCommonOptions.COMPATIBLE_MODE.key(), "mysql");
+ ReadonlyConfig config = ReadonlyConfig.fromMap(options);
+
+ ConfigValidator.of(config).validate(factory.optionRule());
+ ConfigValidator.validateUnknownKeys(config, factory.optionRule(),
"OceanBase-CDC");
+ }
+
+ /** Verify this MySQL-binlog connector rejects the unsupported OceanBase
Oracle mode. */
+ @Test
+ public void testOptionRuleRejectsOracleCompatibleMode() {
+ Map<String, Object> options = requiredOptions();
+ options.put(JdbcCommonOptions.COMPATIBLE_MODE.key(), "oracle");
+
+ Assertions.assertThrows(
+ OptionValidationException.class,
+ () ->
+ ConfigValidator.of(ReadonlyConfig.fromMap(options))
+ .validate(new
OceanBaseIncrementalSourceFactory().optionRule()));
+ }
+
+ private static Map<String, Object> requiredOptions() {
+ Map<String, Object> options = new HashMap<>();
+ options.put("username", "root");
+ options.put("password", "password");
+ options.put("url", "jdbc:mysql://localhost:2881/inventory");
+ options.put("table-names",
Collections.singletonList("inventory.orders"));
+ return options;
+ }
+}
diff --git a/seatunnel-connectors-v2/connector-cdc/pom.xml
b/seatunnel-connectors-v2/connector-cdc/pom.xml
index 1e90099893..f459bc109c 100644
--- a/seatunnel-connectors-v2/connector-cdc/pom.xml
+++ b/seatunnel-connectors-v2/connector-cdc/pom.xml
@@ -33,6 +33,7 @@
<module>connector-cdc-base</module>
<module>connector-cdc-db2</module>
<module>connector-cdc-mysql</module>
+ <module>connector-cdc-oceanbase</module>
<module>connector-cdc-sqlserver</module>
<module>connector-cdc-mongodb</module>
<module>connector-cdc-postgres</module>
diff --git a/seatunnel-dist/pom.xml b/seatunnel-dist/pom.xml
index cdbaf588cb..3b0c38b4ff 100644
--- a/seatunnel-dist/pom.xml
+++ b/seatunnel-dist/pom.xml
@@ -770,6 +770,12 @@
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-cdc-oceanbase</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.apache.seatunnel</groupId>
<artifactId>connector-cdc-oracle</artifactId>
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/pom.xml
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/pom.xml
new file mode 100644
index 0000000000..736fe4bb36
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/pom.xml
@@ -0,0 +1,69 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-connector-v2-e2e</artifactId>
+ <version>${revision}</version>
+ </parent>
+
+ <artifactId>connector-cdc-oceanbase-e2e</artifactId>
+ <name>SeaTunnel : E2E : Connector V2 : CDC OceanBase</name>
+
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-jdbc</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
+ <dependencies>
+ <!-- SeaTunnel connectors -->
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-cdc-base</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-cdc-oceanbase</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>mysql</artifactId>
+ <version>${testcontainer.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>mysql</groupId>
+ <artifactId>mysql-connector-java</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/MySqlCompatibleContainer.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/MySqlCompatibleContainer.java
new file mode 100644
index 0000000000..c3321147eb
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/MySqlCompatibleContainer.java
@@ -0,0 +1,287 @@
+/*
+ * 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.seatunnel.cdc.oceanbase;
+
+import org.testcontainers.containers.ContainerLaunchException;
+import org.testcontainers.containers.JdbcDatabaseContainer;
+import org.testcontainers.utility.DockerImageName;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * MySQL-compatible source container used by the OceanBase CDC E2E module.
+ *
+ * <p>The utility stays local to this module so OceanBase E2E does not depend
on MySQL CDC test
+ * artifacts while still preserving docker-entrypoint initialization for GTID
and binlog settings.
+ */
+class MySqlCompatibleContainer extends
JdbcDatabaseContainer<MySqlCompatibleContainer> {
+
+ /** Official MySQL image repository used as the OceanBase-compatible CDC
runtime. */
+ private static final String IMAGE = "mysql";
+
+ /** MySQL service port exposed inside the Docker network. */
+ private static final Integer MYSQL_PORT = 3306;
+
+ /** Classpath resource parameter that maps a custom my.cnf into the
container. */
+ private static final String MY_CNF_CONFIG_OVERRIDE_PARAM_NAME = "MY_CNF";
+
+ /** Classpath resource parameter that maps setup SQL into
docker-entrypoint-initdb.d. */
+ private static final String SETUP_SQL_PARAM_NAME = "SETUP_SQL";
+
+ /** MySQL root user name, used to decide whether an empty password is
legal. */
+ private static final String MYSQL_ROOT_USER = "root";
+
+ /** Database created by the container entrypoint before setup SQL runs. */
+ private String databaseName = "test";
+
+ /** User created by the container entrypoint for JDBC setup and
assertions. */
+ private String username = "test";
+
+ /** Password used by both the test user and the root account. */
+ private String password = "test";
+
+ /**
+ * Create a MySQL-compatible container with a fixed image tag.
+ *
+ * @param version MySQL image tag used by the E2E test
+ */
+ MySqlCompatibleContainer(String version) {
+ super(DockerImageName.parse(IMAGE + ":" + version));
+ addExposedPort(MYSQL_PORT);
+ }
+
+ /**
+ * Return the mapped MySQL port used by Testcontainers' JDBC wait strategy.
+ *
+ * @return mapped liveness check ports
+ */
+ @Override
+ protected Set<Integer> getLivenessCheckPorts() {
+ Set<Integer> ports = new HashSet<>();
+ ports.add(getMappedPort(MYSQL_PORT));
+ return ports;
+ }
+
+ /** Configure MySQL entrypoint environment and optional classpath-mounted
resources. */
+ @Override
+ protected void configure() {
+ optionallyMapResourceParameterAsVolume(
+ MY_CNF_CONFIG_OVERRIDE_PARAM_NAME, "/etc/mysql/",
"mysql-default-conf");
+
+ if (parameters.containsKey(SETUP_SQL_PARAM_NAME)) {
+ optionallyMapResourceParameterAsVolume(
+ SETUP_SQL_PARAM_NAME, "/docker-entrypoint-initdb.d/",
"N/A");
+ }
+
+ addEnv("MYSQL_DATABASE", databaseName);
+ addEnv("MYSQL_USER", username);
+ if (password != null && !password.isEmpty()) {
+ addEnv("MYSQL_PASSWORD", password);
+ addEnv("MYSQL_ROOT_PASSWORD", password);
+ } else if (MYSQL_ROOT_USER.equalsIgnoreCase(username)) {
+ addEnv("MYSQL_ALLOW_EMPTY_PASSWORD", "yes");
+ } else {
+ throw new ContainerLaunchException(
+ "Empty password can be used only with the root user");
+ }
+ setStartupAttempts(3);
+ }
+
+ /**
+ * Resolve the MySQL driver class name used by the local JDBC assertions.
+ *
+ * @return available MySQL driver class
+ */
+ @Override
+ public String getDriverClassName() {
+ try {
+ Class.forName("com.mysql.cj.jdbc.Driver");
+ return "com.mysql.cj.jdbc.Driver";
+ } catch (ClassNotFoundException e) {
+ return "com.mysql.jdbc.Driver";
+ }
+ }
+
+ /**
+ * Build a JDBC URL for the requested database.
+ *
+ * @param databaseName target database name
+ * @return JDBC URL for the mapped container port
+ */
+ String getJdbcUrl(String databaseName) {
+ String additionalUrlParams = constructUrlParameters("?", "&");
+ return "jdbc:mysql://"
+ + getHost()
+ + ":"
+ + getDatabasePort()
+ + "/"
+ + databaseName
+ + additionalUrlParams;
+ }
+
+ /**
+ * Build the default JDBC URL for this container.
+ *
+ * @return JDBC URL for the configured database
+ */
+ @Override
+ public String getJdbcUrl() {
+ return getJdbcUrl(databaseName);
+ }
+
+ /**
+ * Return the host-mapped MySQL port.
+ *
+ * @return mapped MySQL port
+ */
+ int getDatabasePort() {
+ return getMappedPort(MYSQL_PORT);
+ }
+
+ /**
+ * Add connection parameters needed by MySQL 8 test containers.
+ *
+ * @param queryString query string from Testcontainers
+ * @return JDBC URL with required compatibility parameters
+ */
+ @Override
+ protected String constructUrlForConnection(String queryString) {
+ String url = super.constructUrlForConnection(queryString);
+
+ if (!url.contains("useSSL=")) {
+ url = appendUrlParam(url, "useSSL=false");
+ }
+
+ if (!url.contains("allowPublicKeyRetrieval=")) {
+ url = appendUrlParam(url, "allowPublicKeyRetrieval=true");
+ }
+
+ return url;
+ }
+
+ /**
+ * Return the configured database name.
+ *
+ * @return database name
+ */
+ @Override
+ public String getDatabaseName() {
+ return databaseName;
+ }
+
+ /**
+ * Return the configured JDBC user.
+ *
+ * @return JDBC user name
+ */
+ @Override
+ public String getUsername() {
+ return username;
+ }
+
+ /**
+ * Return the configured JDBC password.
+ *
+ * @return JDBC password
+ */
+ @Override
+ public String getPassword() {
+ return password;
+ }
+
+ /**
+ * Return a cheap query for container readiness checks.
+ *
+ * @return readiness query
+ */
+ @Override
+ protected String getTestQueryString() {
+ return "SELECT 1";
+ }
+
+ /**
+ * Set the classpath my.cnf resource that should be mounted into the
container.
+ *
+ * @param configPath classpath path to the MySQL config file
+ * @return this container
+ */
+ MySqlCompatibleContainer withConfigurationOverride(String configPath) {
+ parameters.put(MY_CNF_CONFIG_OVERRIDE_PARAM_NAME, configPath);
+ return this;
+ }
+
+ /**
+ * Set the classpath setup SQL resource that should run during MySQL
entrypoint initialization.
+ *
+ * @param sqlPath classpath path to setup SQL
+ * @return this container
+ */
+ MySqlCompatibleContainer withSetupSQL(String sqlPath) {
+ parameters.put(SETUP_SQL_PARAM_NAME, sqlPath);
+ return this;
+ }
+
+ /**
+ * Set the database created by the MySQL entrypoint.
+ *
+ * @param databaseName database name
+ * @return this container
+ */
+ @Override
+ public MySqlCompatibleContainer withDatabaseName(final String
databaseName) {
+ this.databaseName = databaseName;
+ return this;
+ }
+
+ /**
+ * Set the JDBC user created by the MySQL entrypoint.
+ *
+ * @param username JDBC user name
+ * @return this container
+ */
+ @Override
+ public MySqlCompatibleContainer withUsername(final String username) {
+ this.username = username;
+ return this;
+ }
+
+ /**
+ * Set the password used by the JDBC and root users.
+ *
+ * @param password JDBC password
+ * @return this container
+ */
+ @Override
+ public MySqlCompatibleContainer withPassword(final String password) {
+ this.password = password;
+ return this;
+ }
+
+ /**
+ * Append a JDBC URL parameter using the correct separator for the current
URL.
+ *
+ * @param url current JDBC URL
+ * @param parameter URL parameter to append
+ * @return JDBC URL with the parameter appended
+ */
+ private String appendUrlParam(String url, String parameter) {
+ String separator = url.contains("?") ? "&" : "?";
+ return url + separator + parameter;
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/OceanBaseCDCCompatibilityIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/OceanBaseCDCCompatibilityIT.java
new file mode 100644
index 0000000000..6a1b2ed3b7
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/OceanBaseCDCCompatibilityIT.java
@@ -0,0 +1,288 @@
+/*
+ * 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.seatunnel.cdc.oceanbase;
+
+import org.apache.seatunnel.e2e.common.TestResource;
+import org.apache.seatunnel.e2e.common.TestSuiteBase;
+import org.apache.seatunnel.e2e.common.container.ContainerExtendedFactory;
+import org.apache.seatunnel.e2e.common.container.EngineType;
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
+import org.apache.seatunnel.e2e.common.junit.TestContainerExtension;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.TestTemplate;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.utility.DockerLoggerFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import static org.awaitility.Awaitility.await;
+
+/**
+ * Verifies that the OceanBase CDC connector can be discovered and run through
the MySQL-compatible
+ * CDC path used by the current OceanBase wrapper.
+ *
+ * <p>This is intentionally a compatibility smoke E2E. It does not claim to
validate a real
+ * OceanBase Binlog Service deployment because the project has no reusable
OceanBase CDC
+ * Testcontainers environment yet.
+ */
+@Slf4j
+@DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK},
+ disabledReason = "Currently SPARK do not support cdc")
+public class OceanBaseCDCCompatibilityIT extends TestSuiteBase implements
TestResource {
+
+ /**
+ * Network alias used by SeaTunnel engine containers to reach the
MySQL-compatible CDC source.
+ */
+ private static final String MYSQL_HOST = "mysql_cdc_e2e";
+
+ /** Administrative user used by the test to prepare source and sink
tables. */
+ private static final String MYSQL_USER_NAME = "mysqluser";
+
+ /** Password shared by the setup SQL users in the MySQL-compatible E2E
environment. */
+ private static final String MYSQL_USER_PASSWORD = "mysqlpw";
+
+ /** Database name shared with the SeaTunnel job config under test. */
+ private static final String MYSQL_DATABASE = "mysql_cdc";
+
+ /** Small source table used to keep the OceanBase wrapper smoke test
focused and fast. */
+ private static final String SOURCE_TABLE =
"oceanbase_cdc_e2e_source_table";
+
+ /** Dedicated sink table generated by the JDBC sink for this compatibility
test. */
+ private static final String SINK_TABLE = "oceanbase_cdc_e2e_sink_table";
+
+ /** MySQL driver jar required by the OceanBase wrapper because it
delegates to MySQL CDC. */
+ private static final String MYSQL_DRIVER_URL =
+
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
+
+ /** MySQL-compatible source container used as the reproducible CDC runtime
for the wrapper. */
+ private static final MySqlCompatibleContainer MYSQL_CONTAINER =
createMySqlContainer();
+
+ /**
+ * Build the MySQL-compatible source with GTID/binlog settings required by
CDC readers.
+ *
+ * @return configured MySQL-compatible container
+ */
+ private static MySqlCompatibleContainer createMySqlContainer() {
+ return new MySqlCompatibleContainer("8.0.43")
+ .withConfigurationOverride("docker/server-gtids/my.cnf")
+ .withSetupSQL("docker/setup.sql")
+ .withNetwork(NETWORK)
+ .withNetworkAliases(MYSQL_HOST)
+ .withDatabaseName(MYSQL_DATABASE)
+ .withUsername(MYSQL_USER_NAME)
+ .withPassword(MYSQL_USER_PASSWORD)
+ .withLogConsumer(
+ new
Slf4jLogConsumer(DockerLoggerFactory.getLogger("mysql-docker-image")));
+ }
+
+ /**
+ * Install the JDBC driver into the OceanBase CDC plugin directory inside
each engine container.
+ */
+ @TestContainerExtension
+ protected final ContainerExtendedFactory extendedFactory =
+ container -> {
+ Container.ExecResult extraCommands =
+ container.execInContainer(
+ "bash",
+ "-c",
+ "mkdir -p
/tmp/seatunnel/plugins/OceanBase-CDC/lib && cd
/tmp/seatunnel/plugins/OceanBase-CDC/lib && wget "
+ + MYSQL_DRIVER_URL);
+ Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
+ };
+
+ /** Start the MySQL-compatible CDC source before the SeaTunnel job
containers run. */
+ @BeforeAll
+ @Override
+ public void startUp() {
+ log.info("Starting MySQL-compatible container for OceanBase CDC
compatibility E2E.");
+ Startables.deepStart(Stream.of(MYSQL_CONTAINER)).join();
+ log.info("MySQL-compatible container for OceanBase CDC compatibility
E2E is started.");
+ }
+
+ /** Close the MySQL container allocated by this test class. */
+ @Override
+ @AfterAll
+ public void tearDown() {
+ if (MYSQL_CONTAINER != null) {
+ MYSQL_CONTAINER.close();
+ }
+ }
+
+ /**
+ * Run OceanBase-CDC against a MySQL-compatible CDC source and verify
snapshot plus incremental
+ * events reach the JDBC sink.
+ *
+ * @param container SeaTunnel engine test container
+ */
+ @TestTemplate
+ public void testOceanBaseCdcWrapperRuntimeE2e(TestContainer container) {
+ recreateTestTables();
+
+ CompletableFuture<Void> jobFuture =
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ Container.ExecResult execResult =
+ container.executeJob(
+
"/oceanbasecdc_to_mysql_compatible.conf");
+ Assertions.assertEquals(
+ 0, execResult.getExitCode(),
execResult.getStderr());
+ } catch (Exception e) {
+ throw new CompletionException(e);
+ }
+ });
+
+ await().atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ assertStreamingJobStillRunning(jobFuture);
+ Assertions.assertIterableEquals(
+ queryTable(SOURCE_TABLE),
queryTable(SINK_TABLE));
+ });
+
+ applyIncrementalChanges();
+
+ await().atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ assertStreamingJobStillRunning(jobFuture);
+ Assertions.assertIterableEquals(
+ queryTable(SOURCE_TABLE),
queryTable(SINK_TABLE));
+ });
+ }
+
+ /** Recreate isolated source and sink tables for deterministic snapshot
assertions. */
+ private void recreateTestTables() {
+ executeSql("DROP TABLE IF EXISTS " + MYSQL_DATABASE + "." +
SINK_TABLE);
+ executeSql("DROP TABLE IF EXISTS " + MYSQL_DATABASE + "." +
SOURCE_TABLE);
+ executeSql(
+ "CREATE TABLE "
+ + MYSQL_DATABASE
+ + "."
+ + SOURCE_TABLE
+ + " (id INT PRIMARY KEY, name VARCHAR(64), score
INT)");
+ executeSql(
+ "CREATE TABLE "
+ + MYSQL_DATABASE
+ + "."
+ + SINK_TABLE
+ + " (id INT PRIMARY KEY, name VARCHAR(64), score
INT)");
+ executeSql(
+ "INSERT INTO "
+ + MYSQL_DATABASE
+ + "."
+ + SOURCE_TABLE
+ + " VALUES (1, 'alpha', 10), (2, 'beta', 20)");
+ }
+
+ /** Apply insert, update and delete operations after the initial snapshot
is synchronized. */
+ private void applyIncrementalChanges() {
+ executeSql(
+ "INSERT INTO " + MYSQL_DATABASE + "." + SOURCE_TABLE + "
VALUES (3, 'gamma', 30)");
+ executeSql(
+ "UPDATE " + MYSQL_DATABASE + "." + SOURCE_TABLE + " SET score
= 25 WHERE id = 2");
+ executeSql("DELETE FROM " + MYSQL_DATABASE + "." + SOURCE_TABLE + "
WHERE id = 1");
+ }
+
+ /**
+ * Fail fast if the asynchronous streaming job has already exited or
failed.
+ *
+ * @param jobFuture asynchronous SeaTunnel job execution future
+ */
+ private void assertStreamingJobStillRunning(CompletableFuture<Void>
jobFuture) {
+ if (jobFuture.isCompletedExceptionally()) {
+ jobFuture.join();
+ }
+ Assertions.assertFalse(jobFuture.isDone(), "OceanBase CDC streaming
job exited early.");
+ }
+
+ /**
+ * Query the deterministic table projection used by this test.
+ *
+ * @param tableName source or sink table name
+ * @return rows ordered by primary key
+ */
+ private List<List<Object>> queryTable(String tableName) {
+ String sql =
+ "SELECT id, name, score FROM " + MYSQL_DATABASE + "." +
tableName + " ORDER BY id";
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery(sql)) {
+ List<List<Object>> result = new ArrayList<>();
+ while (resultSet.next()) {
+ result.add(
+ Arrays.asList(
+ resultSet.getInt("id"),
+ resultSet.getString("name"),
+ resultSet.getInt("score")));
+ }
+ log.debug("OceanBase CDC compatibility query, sql: {}, data: {}",
sql, result);
+ return result;
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Execute table setup and mutation SQL against the MySQL-compatible CDC
source.
+ *
+ * @param sql SQL statement to execute
+ */
+ private void executeSql(String sql) {
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(sql);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Open an administrative JDBC connection to the MySQL-compatible source
container.
+ *
+ * @return JDBC connection for setup and assertions
+ * @throws SQLException when the container rejects the connection
+ */
+ private Connection getJdbcConnection() throws SQLException {
+ return DriverManager.getConnection(
+ MYSQL_CONTAINER.getJdbcUrl(),
+ MYSQL_CONTAINER.getUsername(),
+ MYSQL_CONTAINER.getPassword());
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/src/test/resources/docker/server-gtids/my.cnf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/src/test/resources/docker/server-gtids/my.cnf
new file mode 100644
index 0000000000..bfe5641b9c
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/src/test/resources/docker/server-gtids/my.cnf
@@ -0,0 +1,31 @@
+#
+# 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.
+#
+
+[mysqld]
+skip-host-cache
+skip-name-resolve
+secure-file-priv=/var/lib/mysql
+user=mysql
+symbolic-links=0
+
+server-id = 223344
+log_bin = mysql-bin
+expire_logs_days = 1
+binlog_format = row
+
+gtid_mode = on
+enforce_gtid_consistency = on
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/src/test/resources/docker/setup.sql
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/src/test/resources/docker/setup.sql
new file mode 100644
index 0000000000..26eae4ac5c
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/src/test/resources/docker/setup.sql
@@ -0,0 +1,24 @@
+--
+-- 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.
+--
+
+GRANT ALL PRIVILEGES ON *.* TO 'mysqluser'@'%';
+
+CREATE USER 'st_user_source' IDENTIFIED BY 'mysqlpw';
+GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT,
DROP, LOCK TABLES ON *.* TO 'st_user_source'@'%';
+
+CREATE USER 'st_user_sink' IDENTIFIED BY 'mysqlpw';
+GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, DROP, INDEX, ALTER ON *.* TO
'st_user_sink'@'%';
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/src/test/resources/oceanbasecdc_to_mysql_compatible.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/src/test/resources/oceanbasecdc_to_mysql_compatible.conf
new file mode 100644
index 0000000000..0315ee7e81
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oceanbase-e2e/src/test/resources/oceanbasecdc_to_mysql_compatible.conf
@@ -0,0 +1,49 @@
+#
+# 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.
+#
+
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ OceanBase-CDC {
+ plugin_output = "customers_oceanbase_cdc"
+ server-id = 5752
+ username = "st_user_source"
+ password = "mysqlpw"
+ compatible_mode = "mysql"
+ table-names = ["mysql_cdc.oceanbase_cdc_e2e_source_table"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ }
+}
+
+sink {
+ jdbc {
+ plugin_input = "customers_oceanbase_cdc"
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "st_user_sink"
+ password = "mysqlpw"
+
+ generate_sink_sql = true
+ database = mysql_cdc
+ table = oceanbase_cdc_e2e_sink_table
+ primary_keys = ["id"]
+ }
+}
diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml
b/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml
index 4dc825191b..9a488d3b5b 100644
--- a/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml
+++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml
@@ -61,6 +61,7 @@
<module>connector-iotdb-e2e</module>
<module>connector-iotdb-v2-e2e</module>
<module>connector-cdc-mysql-e2e</module>
+ <module>connector-cdc-oceanbase-e2e</module>
<module>connector-cdc-mongodb-e2e</module>
<module>connector-iceberg-e2e</module>
<module>connector-iceberg-hadoop3-e2e</module>
