This is an automated email from the ASF dual-hosted git repository.
JNSimba pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 8a76160b646 [feature](streaming-job) Support OceanBase CDC streaming
jobs (#65588)
8a76160b646 is described below
commit 8a76160b6460fbf8c406c3e6b770142e43719200
Author: wudi <[email protected]>
AuthorDate: Thu Jul 23 18:50:43 2026 +0800
[feature](streaming-job) Support OceanBase CDC streaming jobs (#65588)
### What problem does this PR solve?
Related PR: #65325
Streaming jobs currently support MySQL and PostgreSQL CDC sources but
cannot use OceanBase as a source.
This change adds OceanBase MySQL compatibility mode as a streaming job
data source. It introduces OceanBase source type handling in FE and the
CDC client, reuses the MySQL-compatible CDC processing path, validates
source configuration and compatibility mode before creating target
tables, and supports initial, snapshot, latest, earliest, and specific
startup offsets.
The OceanBase third-party environment exposes separate JDBC and CDC
ports. Test coverage includes basic synchronization, data types, startup
offsets, schema changes, and FE restart recovery.
---
.../docker-compose/oceanbase/oceanbase.env | 1 +
.../docker-compose/oceanbase/oceanbase.yaml.tpl | 6 +-
.../apache/doris/job/common/DataSourceType.java | 3 +-
.../streaming/DataSourceConfigValidator.java | 79 +++++++-
.../insert/streaming/StreamingInsertJob.java | 19 +-
.../streaming/StreamingJdbcUrlNormalizer.java | 1 +
.../apache/doris/job/util/StreamingJobUtils.java | 15 +-
.../streaming/DataSourceConfigValidatorTest.java | 188 +++++++++++++++++++
.../streaming/StreamingJdbcUrlNormalizerTest.java | 10 +
.../doris/job/util/StreamingJobUtilsTest.java | 13 ++
.../mysql/MySqlStreamingChangeEventSource.java | 92 ++++++----
.../doris/cdcclient/source/factory/DataSource.java | 3 +-
.../source/factory/SourceReaderFactory.java | 2 +
.../source/reader/mysql/MySqlSourceReader.java | 9 +
.../reader/oceanbase/OceanBaseSourceReader.java | 15 +-
.../mysql/MySqlStreamingChangeEventSourceTest.java | 48 +++++
.../cdcclient/itcase/CdcClientWriteHarness.java | 53 +++++-
.../itcase/OceanBaseSchemaChangeITCase.java | 121 ++++++++++++
.../itcase/OceanBaseStartupOffsetITCase.java | 148 +++++++++++++++
.../doris/cdcclient/itcase/OceanBaseTestBase.java | 150 +++++++++++++++
.../cdcclient/itcase/OceanBaseWriteDmlITCase.java | 112 ++++++++++++
.../source/factory/SourceReaderFactoryTest.java | 9 +
.../source/reader/mysql/MySqlSourceReaderTest.java | 31 +++-
regression-test/conf/regression-conf.groovy | 3 +-
.../cdc/test_streaming_oceanbase_job.out | 17 ++
.../cdc/test_streaming_oceanbase_job_all_type.out | 8 +
.../cdc/test_streaming_oceanbase_job_offset.out | 11 ++
.../test_streaming_oceanbase_job_restart_fe.out | 5 +
.../test_streaming_oceanbase_job_sc_restart_fe.out | 11 ++
.../test_streaming_oceanbase_job_schema_change.out | 9 +
.../pipeline/external/conf/regression-conf.groovy | 1 +
.../suites/doc/external/jdbc/oceanbase.md.groovy | 2 +-
.../jdbc/test_oceanbase_jdbc_catalog.groovy | 4 +-
.../cdc/test_streaming_oceanbase_job.groovy | 143 +++++++++++++++
.../test_streaming_oceanbase_job_all_type.groovy | 152 +++++++++++++++
.../cdc/test_streaming_oceanbase_job_offset.groovy | 203 +++++++++++++++++++++
.../test_streaming_oceanbase_job_restart_fe.groovy | 150 +++++++++++++++
...st_streaming_oceanbase_job_sc_restart_fe.groovy | 156 ++++++++++++++++
...st_streaming_oceanbase_job_schema_change.groovy | 138 ++++++++++++++
39 files changed, 2071 insertions(+), 70 deletions(-)
diff --git a/docker/thirdparties/docker-compose/oceanbase/oceanbase.env
b/docker/thirdparties/docker-compose/oceanbase/oceanbase.env
index 5a8998e7cd5..108843f9686 100644
--- a/docker/thirdparties/docker-compose/oceanbase/oceanbase.env
+++ b/docker/thirdparties/docker-compose/oceanbase/oceanbase.env
@@ -17,3 +17,4 @@
# under the License.
DOCKER_OCEANBASE_EXTERNAL_PORT=2881
+DOCKER_OCEANBASE_PROXY_EXTERNAL_PORT=2883
diff --git a/docker/thirdparties/docker-compose/oceanbase/oceanbase.yaml.tpl
b/docker/thirdparties/docker-compose/oceanbase/oceanbase.yaml.tpl
index aa64c1d2144..fee16e3c160 100644
--- a/docker/thirdparties/docker-compose/oceanbase/oceanbase.yaml.tpl
+++ b/docker/thirdparties/docker-compose/oceanbase/oceanbase.yaml.tpl
@@ -19,16 +19,18 @@ version: "2.1"
services:
doris--oceanbase:
- image: oceanbase/oceanbase-ce:4.2.1-lts
+ image: quay.io/oceanbase/obbinlog-ce:4.2.5-test
restart: always
environment:
MODE: slim
OB_MEMORY_LIMIT: 5G
+ PASSWORD: 123456
TZ: Asia/Shanghai
ports:
- ${DOCKER_OCEANBASE_EXTERNAL_PORT}:2881
+ - ${DOCKER_OCEANBASE_PROXY_EXTERNAL_PORT}:2883
healthcheck:
- test: ["CMD-SHELL", "obclient -h127.1 -P2881 -uroot@test -e 'SELECT *
FROM doris_test.all_types limit 1'"]
+ test: ["CMD-SHELL", "obclient -h127.0.0.1 -P2881 -uroot@test -p123456 -e
'SELECT * FROM doris_test.all_types LIMIT 1' >/dev/null && obclient -h127.0.0.1
-P2883 -uroot@test -p123456 -Nse 'SHOW MASTER STATUS' | grep -q ."]
interval: 5s
timeout: 60s
retries: 120
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/common/DataSourceType.java
b/fe/fe-core/src/main/java/org/apache/doris/job/common/DataSourceType.java
index b188f265957..3eec77e757b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/common/DataSourceType.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/common/DataSourceType.java
@@ -19,5 +19,6 @@ package org.apache.doris.job.common;
public enum DataSourceType {
MYSQL,
- POSTGRES
+ POSTGRES,
+ OCEANBASE
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidator.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidator.java
index 1c1dc10c0d5..f75bf03c56e 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidator.java
@@ -17,8 +17,12 @@
package org.apache.doris.job.extensions.insert.streaming;
+import org.apache.doris.datasource.jdbc.client.JdbcClient;
+import org.apache.doris.datasource.jdbc.client.JdbcClientException;
import org.apache.doris.job.cdc.DataSourceConfigKeys;
import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.util.StreamingJobUtils;
import org.apache.doris.nereids.trees.plans.commands.LoadCommand;
import com.fasterxml.jackson.databind.JsonNode;
@@ -26,6 +30,9 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableSortedSet;
import com.google.common.collect.Sets;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;
@@ -36,6 +43,7 @@ public class DataSourceConfigValidator {
// PostgreSQL unquoted identifier: lowercase letters, digits, underscores,
not starting with a digit.
private static final Pattern PG_IDENTIFIER_PATTERN =
Pattern.compile("^[a-z_][a-z0-9_]*$");
private static final int PG_MAX_IDENTIFIER_LENGTH = 63;
+ private static final String MYSQL_JDBC_URL_PREFIX = "jdbc:mysql://";
private static final Set<String> ALLOW_SOURCE_KEYS = Sets.newHashSet(
DataSourceConfigKeys.JDBC_URL,
@@ -58,6 +66,12 @@ public class DataSourceConfigValidator {
DataSourceConfigKeys.SERVER_ID
);
+ private static final Set<String> OCEANBASE_UNSUPPORTED_KEYS =
Sets.newHashSet(
+ DataSourceConfigKeys.SCHEMA,
+ DataSourceConfigKeys.SLOT_NAME,
+ DataSourceConfigKeys.PUBLICATION_NAME
+ );
+
private static final Set<String> ALLOW_SSL_MODES = Sets.newHashSet(
DataSourceConfigKeys.SSL_MODE_DISABLE,
DataSourceConfigKeys.SSL_MODE_REQUIRE,
@@ -106,14 +120,72 @@ public class DataSourceConfigValidator {
throw new IllegalArgumentException("Unexpected key: '" + key +
"'");
}
+ if
(DataSourceType.OCEANBASE.name().equalsIgnoreCase(dataSourceType)
+ && OCEANBASE_UNSUPPORTED_KEYS.contains(key)) {
+ throw new IllegalArgumentException(
+ "Property '" + key + "' is not supported for
OceanBase");
+ }
+
if (!isValidValue(key, value, dataSourceType)) {
throw new IllegalArgumentException("Invalid value for key '" +
key + "': " + value);
}
}
+ validateOceanBaseSource(input, dataSourceType);
validateSslVerifyCaPair(input);
}
+ private static void validateOceanBaseSource(Map<String, String> input,
String dataSourceType) {
+ if (!DataSourceType.OCEANBASE.name().equalsIgnoreCase(dataSourceType))
{
+ return;
+ }
+ if (input.containsKey(DataSourceConfigKeys.JDBC_URL)
+ &&
!input.get(DataSourceConfigKeys.JDBC_URL).startsWith(MYSQL_JDBC_URL_PREFIX)) {
+ throw new IllegalArgumentException(
+ "OceanBase jdbc_url must start with '" +
MYSQL_JDBC_URL_PREFIX + "'");
+ }
+ }
+
+ public static void validateSourceBeforeTableCreation(
+ DataSourceType sourceType, Map<String, String> sourceProperties)
throws JobException {
+ if (sourceType == DataSourceType.OCEANBASE) {
+ validateOceanBaseCompatibilityMode(sourceProperties);
+ }
+ }
+
+ private static void validateOceanBaseCompatibilityMode(Map<String, String>
sourceProperties)
+ throws JobException {
+ // jdbc:mysql routes through JdbcMySQLClient so Connector/J is
initialized consistently.
+ JdbcClient jdbcClient = StreamingJobUtils.getJdbcClient(
+ DataSourceType.OCEANBASE, sourceProperties);
+ try (Connection connection = jdbcClient.getConnection();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery(
+ "SHOW VARIABLES LIKE 'ob_compatibility_mode'")) {
+ if (!resultSet.next()) {
+ throw new JobException("Failed to determine OceanBase
compatibility mode");
+ }
+ String compatibilityMode = resultSet.getString(2);
+ if ("MYSQL".equalsIgnoreCase(compatibilityMode)) {
+ return;
+ }
+ if ("ORACLE".equalsIgnoreCase(compatibilityMode)) {
+ throw new JobException(
+ "OceanBase Oracle compatibility mode is not supported
for streaming jobs");
+ }
+ throw new JobException(
+ "Unsupported OceanBase compatibility mode: " +
compatibilityMode);
+ } catch (JobException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new JobException(
+ "Failed to validate OceanBase compatibility mode: "
+ + JdbcClientException.getAllExceptionMessages(e),
e);
+ } finally {
+ jdbcClient.closeClient();
+ }
+ }
+
// Cross-field: verify-ca must be paired with a CA cert; otherwise the
reader will
// silently fall back to the JVM default truststore and likely fail to
connect.
public static void validateSslVerifyCaPair(Map<String, String> input)
throws IllegalArgumentException {
@@ -294,7 +366,7 @@ public class DataSourceConfigValidator {
/**
* Check if the offset value is valid for the given data source type.
* Supported: initial, snapshot, latest, JSON binlog/lsn position.
- * earliest is only supported for MySQL.
+ * earliest is only supported for MySQL-compatible sources.
*/
public static boolean isValidOffset(String offset, String dataSourceType) {
if (offset == null || offset.isEmpty()) {
@@ -305,9 +377,10 @@ public class DataSourceConfigValidator {
||
DataSourceConfigKeys.OFFSET_SNAPSHOT.equalsIgnoreCase(offset)) {
return true;
}
- // earliest only for MySQL
+ // earliest only for MySQL-compatible sources
if (DataSourceConfigKeys.OFFSET_EARLIEST.equalsIgnoreCase(offset)) {
- return
DataSourceType.MYSQL.name().equalsIgnoreCase(dataSourceType);
+ return DataSourceType.MYSQL.name().equalsIgnoreCase(dataSourceType)
+ ||
DataSourceType.OCEANBASE.name().equalsIgnoreCase(dataSourceType);
}
if (isJsonOffset(offset)) {
return true;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
index 28b684a7e03..cc7e7e34a18 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
@@ -247,9 +247,9 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
}
/**
- * Initialize job from source to database, like multi table mysql to doris.
- * 1. get mysql connection info from sourceProperties
- * 2. fetch table list from mysql
+ * Initialize a multi-table job from an external database source to Doris.
+ * 1. get source connection info from sourceProperties
+ * 2. fetch the source table list
* 3. create doris table if not exists
* 4. check whether need full data sync
* 5. need => fetch split and write to system table
@@ -258,6 +258,8 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
try {
init();
checkRequiredSourceProperties();
+ DataSourceConfigValidator.validateSourceBeforeTableCreation(
+ dataSourceType, sourceProperties);
List<String> createTbls = createTableIfNotExists();
this.syncTables = createTbls;
if (sourceProperties.get(DataSourceConfigKeys.INCLUDE_TABLES) ==
null) {
@@ -271,7 +273,7 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
this.offsetProvider.initOnCreate(this.syncTables);
} catch (Exception ex) {
log.warn("init streaming job for {} failed", dataSourceType, ex);
- throw new RuntimeException(ex.getMessage());
+ throw new RuntimeException(ex.getMessage(), ex);
}
}
@@ -298,7 +300,7 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
private List<String> createTableIfNotExists() throws Exception {
List<String> syncTbls = new ArrayList<>();
Map<String, String> effectiveSourceProperties =
buildConvertedSourceProperties(sourceProperties);
- // Key: source table name (PG/MySQL); Value: CreateTableCommand for
the Doris target table.
+ // Key: source table name; Value: CreateTableCommand for the Doris
target table.
// The two names differ when "table.<src>.target_table" is configured.
LinkedHashMap<String, CreateTableCommand> createTblCmds =
StreamingJobUtils.generateCreateTableCmds(targetDb,
@@ -311,7 +313,7 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
if
(!db.isTableExist(createTblCmd.getCreateTableInfo().getTableName())) {
createTblCmd.run(ConnectContext.get(), null);
}
- // Use the source (upstream) table name so CDC monitors the
correct PG/MySQL table
+ // Use the upstream table name so CDC monitors the correct source
table.
syncTbls.add(srcTable);
}
return syncTbls;
@@ -654,10 +656,7 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
return runningStreamTask;
}
- /**
- * for From MySQL TO Database
- * @return
- */
+ /** Create a task for a FROM source TO DATABASE streaming job. */
private AbstractStreamingTask createStreamingMultiTblTask() throws
JobException {
return new StreamingMultiTblTask(getJobId(),
Env.getCurrentEnv().getNextId(), dataSourceType,
offsetProvider, getConvertedSourceProperties(), targetDb,
targetProperties, jobProperties,
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizer.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizer.java
index f13d41687df..70f776ab5c1 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizer.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizer.java
@@ -35,6 +35,7 @@ public final class StreamingJdbcUrlNormalizer {
public static String normalize(DataSourceType sourceType, String jdbcUrl) {
switch (sourceType) {
case MYSQL:
+ case OCEANBASE:
return normalizeMysql(jdbcUrl);
case POSTGRES:
return jdbcUrl;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java
b/fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java
index 033cc4404bb..1ef6bf18093 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java
@@ -359,7 +359,7 @@ public class StreamingJobUtils {
* <p>Returns a {@link LinkedHashMap} whose key is the <b>source</b>
(upstream) table name and
* whose value is the corresponding {@link CreateTableCommand} that
creates the Doris target
* table (which may have a different name when {@code
table.<src>.target_table} is configured).
- * Callers must use the map key as the PG/MySQL source table identifier
for CDC monitoring and
+ * Callers must use the map key as the upstream source table identifier
for CDC monitoring and
* the {@link CreateTableCommand} value for the actual DDL execution.
*/
public static LinkedHashMap<String, CreateTableCommand>
generateCreateTableCmds(String targetDb,
@@ -520,14 +520,6 @@ public class StreamingJobUtils {
return columns;
}
- /**
- * The remoteDB implementation differs for each data source;
- * refer to the hierarchical mapping in the JDBC catalog.
- */
- /**
- * Populate default resource names into properties, then validate. No-op
for sources that
- * don't need it. Mutates properties: callers should expect default values
to be inserted.
- */
public static void resolveAndValidateSource(DataSourceType sourceType,
Map<String, String> properties,
String jobId,
@@ -591,10 +583,15 @@ public class StreamingJobUtils {
}
}
+ /**
+ * The remoteDB implementation differs for each data source;
+ * refer to the hierarchical mapping in the JDBC catalog.
+ */
public static String getRemoteDbName(DataSourceType sourceType,
Map<String, String> properties) {
String remoteDb = null;
switch (sourceType) {
case MYSQL:
+ case OCEANBASE:
remoteDb = properties.get(DataSourceConfigKeys.DATABASE);
Preconditions.checkArgument(StringUtils.isNotEmpty(remoteDb),
"database is required");
break;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidatorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidatorTest.java
index 84cc3bdf5e0..3765fd10189 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidatorTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidatorTest.java
@@ -17,12 +17,20 @@
package org.apache.doris.job.extensions.insert.streaming;
+import org.apache.doris.datasource.jdbc.client.JdbcClient;
import org.apache.doris.job.cdc.DataSourceConfigKeys;
import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.util.StreamingJobUtils;
import org.junit.Assert;
import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
import java.util.HashMap;
import java.util.Map;
@@ -398,4 +406,184 @@ public class DataSourceConfigValidatorTest {
props.put(DataSourceConfigKeys.SNAPSHOT_PARALLELISM, "4");
DataSourceConfigValidator.validateSource(props,
DataSourceType.MYSQL.name());
}
+
+ @Test
+ public void testOceanBaseAcceptsMysqlJdbcUrl() {
+ Map<String, String> props = new HashMap<>();
+ props.put(DataSourceConfigKeys.JDBC_URL,
"jdbc:mysql://localhost:2883/test_db");
+
+ DataSourceConfigValidator.validateSource(props,
DataSourceType.OCEANBASE.name());
+ }
+
+ @Test
+ public void testOceanBaseAllowsPartialPropertiesWithoutJdbcUrl() {
+ Map<String, String> props = new HashMap<>();
+ props.put(DataSourceConfigKeys.SNAPSHOT_PARALLELISM, "2");
+
+ DataSourceConfigValidator.validateSource(props,
DataSourceType.OCEANBASE.name());
+ }
+
+ @Test
+ public void testOceanBaseRejectsNonMysqlJdbcUrl() {
+ Map<String, String> props = new HashMap<>();
+ props.put(DataSourceConfigKeys.JDBC_URL,
"jdbc:oceanbase://localhost:2883/test_db");
+
+ IllegalArgumentException exception =
Assert.assertThrows(IllegalArgumentException.class,
+ () -> DataSourceConfigValidator.validateSource(
+ props, DataSourceType.OCEANBASE.name()));
+
+ Assert.assertTrue(exception.getMessage().contains("jdbc:mysql://"));
+ }
+
+ @Test
+ public void testOceanBaseRejectsPostgresProperties() {
+ for (String key : new String[] {
+ DataSourceConfigKeys.SCHEMA,
+ DataSourceConfigKeys.SLOT_NAME,
+ DataSourceConfigKeys.PUBLICATION_NAME}) {
+ Map<String, String> props = new HashMap<>();
+ props.put(key, "value");
+
+ IllegalArgumentException exception = Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> DataSourceConfigValidator.validateSource(
+ props, DataSourceType.OCEANBASE.name()));
+ Assert.assertTrue(exception.getMessage().contains(key));
+ }
+ }
+
+ @Test
+ public void testOceanBaseDoesNotExposeSchemaChangeEnabled() {
+ Map<String, String> props = new HashMap<>();
+ props.put(DataSourceConfigKeys.SCHEMA_CHANGE_ENABLED, "false");
+
+ IllegalArgumentException exception =
Assert.assertThrows(IllegalArgumentException.class,
+ () -> DataSourceConfigValidator.validateSource(
+ props, DataSourceType.OCEANBASE.name()));
+
+
Assert.assertTrue(exception.getMessage().contains(DataSourceConfigKeys.SCHEMA_CHANGE_ENABLED));
+ }
+
+ @Test
+ public void testOceanBaseSupportsEarliestOffset() {
+ Assert.assertTrue(DataSourceConfigValidator.isValidOffset(
+ DataSourceConfigKeys.OFFSET_EARLIEST,
DataSourceType.OCEANBASE.name()));
+ Assert.assertFalse(DataSourceConfigValidator.isValidOffset(
+ DataSourceConfigKeys.OFFSET_EARLIEST,
DataSourceType.POSTGRES.name()));
+ }
+
+ @Test
+ public void testCompatibilityModePreflightIsNoopForExistingSources()
throws Exception {
+ DataSourceConfigValidator.validateSourceBeforeTableCreation(
+ DataSourceType.MYSQL, new HashMap<>());
+ DataSourceConfigValidator.validateSourceBeforeTableCreation(
+ DataSourceType.POSTGRES, new HashMap<>());
+ }
+
+ @Test
+ public void testOceanBaseMysqlCompatibilityModePasses() throws Exception {
+ JdbcClient jdbcClient = mockOceanBaseCompatibilityMode(true, "MYSQL");
+
+ try (MockedStatic<StreamingJobUtils> utils =
Mockito.mockStatic(StreamingJobUtils.class)) {
+ utils.when(() -> StreamingJobUtils.getJdbcClient(
+ Mockito.eq(DataSourceType.OCEANBASE),
Mockito.anyMap()))
+ .thenReturn(jdbcClient);
+
+ DataSourceConfigValidator.validateSourceBeforeTableCreation(
+ DataSourceType.OCEANBASE, new HashMap<>());
+ }
+
+ Mockito.verify(jdbcClient).closeClient();
+ }
+
+ @Test
+ public void testOceanBaseOracleCompatibilityModeIsRejected() throws
Exception {
+ JdbcClient jdbcClient = mockOceanBaseCompatibilityMode(true, "ORACLE");
+
+ try (MockedStatic<StreamingJobUtils> utils =
Mockito.mockStatic(StreamingJobUtils.class)) {
+ utils.when(() -> StreamingJobUtils.getJdbcClient(
+ Mockito.eq(DataSourceType.OCEANBASE),
Mockito.anyMap()))
+ .thenReturn(jdbcClient);
+
+ JobException exception = Assert.assertThrows(JobException.class,
+ () ->
DataSourceConfigValidator.validateSourceBeforeTableCreation(
+ DataSourceType.OCEANBASE, new HashMap<>()));
+ Assert.assertTrue(exception.getMessage().contains("Oracle
compatibility mode"));
+ }
+
+ Mockito.verify(jdbcClient).closeClient();
+ }
+
+ @Test
+ public void testOceanBaseUnknownCompatibilityModeIsRejected() throws
Exception {
+ JdbcClient jdbcClient = mockOceanBaseCompatibilityMode(true,
"UNKNOWN");
+
+ try (MockedStatic<StreamingJobUtils> utils =
Mockito.mockStatic(StreamingJobUtils.class)) {
+ utils.when(() -> StreamingJobUtils.getJdbcClient(
+ Mockito.eq(DataSourceType.OCEANBASE),
Mockito.anyMap()))
+ .thenReturn(jdbcClient);
+
+ JobException exception = Assert.assertThrows(JobException.class,
+ () ->
DataSourceConfigValidator.validateSourceBeforeTableCreation(
+ DataSourceType.OCEANBASE, new HashMap<>()));
+ Assert.assertTrue(exception.getMessage().contains("UNKNOWN"));
+ }
+
+ Mockito.verify(jdbcClient).closeClient();
+ }
+
+ @Test
+ public void testOceanBaseEmptyCompatibilityModeResultIsRejected() throws
Exception {
+ JdbcClient jdbcClient = mockOceanBaseCompatibilityMode(false, null);
+
+ try (MockedStatic<StreamingJobUtils> utils =
Mockito.mockStatic(StreamingJobUtils.class)) {
+ utils.when(() -> StreamingJobUtils.getJdbcClient(
+ Mockito.eq(DataSourceType.OCEANBASE),
Mockito.anyMap()))
+ .thenReturn(jdbcClient);
+
+ JobException exception = Assert.assertThrows(JobException.class,
+ () ->
DataSourceConfigValidator.validateSourceBeforeTableCreation(
+ DataSourceType.OCEANBASE, new HashMap<>()));
+ Assert.assertTrue(exception.getMessage().contains("Failed to
determine"));
+ }
+
+ Mockito.verify(jdbcClient).closeClient();
+ }
+
+ @Test
+ public void testOceanBaseCompatibilityModeQueryFailurePreservesCause()
throws Exception {
+ JdbcClient jdbcClient = Mockito.mock(JdbcClient.class);
+ Connection connection = Mockito.mock(Connection.class);
+ Mockito.when(jdbcClient.getConnection()).thenReturn(connection);
+ Mockito.when(connection.createStatement()).thenThrow(new
IllegalStateException("query failed"));
+
+ try (MockedStatic<StreamingJobUtils> utils =
Mockito.mockStatic(StreamingJobUtils.class)) {
+ utils.when(() -> StreamingJobUtils.getJdbcClient(
+ Mockito.eq(DataSourceType.OCEANBASE),
Mockito.anyMap()))
+ .thenReturn(jdbcClient);
+
+ JobException exception = Assert.assertThrows(JobException.class,
+ () ->
DataSourceConfigValidator.validateSourceBeforeTableCreation(
+ DataSourceType.OCEANBASE, new HashMap<>()));
+ Assert.assertTrue(exception.getMessage().contains("query failed"));
+ Assert.assertNotNull(exception.getCause());
+ }
+
+ Mockito.verify(jdbcClient).closeClient();
+ }
+
+ private JdbcClient mockOceanBaseCompatibilityMode(boolean hasResult,
String mode)
+ throws Exception {
+ JdbcClient jdbcClient = Mockito.mock(JdbcClient.class);
+ Connection connection = Mockito.mock(Connection.class);
+ Statement statement = Mockito.mock(Statement.class);
+ ResultSet resultSet = Mockito.mock(ResultSet.class);
+ Mockito.when(jdbcClient.getConnection()).thenReturn(connection);
+ Mockito.when(connection.createStatement()).thenReturn(statement);
+ Mockito.when(statement.executeQuery("SHOW VARIABLES LIKE
'ob_compatibility_mode'"))
+ .thenReturn(resultSet);
+ Mockito.when(resultSet.next()).thenReturn(hasResult);
+ Mockito.when(resultSet.getString(2)).thenReturn(mode);
+ return jdbcClient;
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizerTest.java
index bfdc7ec695d..2dda159a11d 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizerTest.java
@@ -68,6 +68,16 @@ public class StreamingJdbcUrlNormalizerTest {
StreamingJdbcUrlNormalizer.normalize(DataSourceType.MYSQL,
jdbcUrl));
}
+ @Test
+ public void testNormalizeOceanBaseJdbcUrlUsesMysqlRules() {
+ String jdbcUrl = StreamingJdbcUrlNormalizer.normalize(
+ DataSourceType.OCEANBASE, "jdbc:mysql://127.0.0.1:2883/test");
+
+
Assert.assertEquals("jdbc:mysql://127.0.0.1:2883/test?yearIsDateType=false"
+ +
"&tinyInt1isBit=false&useUnicode=true&characterEncoding=utf-8",
+ jdbcUrl);
+ }
+
@Test
public void testNormalizePostgresJdbcUrlDoesNotChange() {
String jdbcUrl = "jdbc:postgresql://127.0.0.1:5432/test";
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java
index 3e090d4d3fe..93c19074cd1 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java
@@ -21,6 +21,8 @@ import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.PrimitiveType;
import org.apache.doris.catalog.ScalarType;
import org.apache.doris.datasource.jdbc.client.JdbcClient;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.common.DataSourceType;
import org.junit.Assert;
import org.junit.Before;
@@ -32,7 +34,9 @@ import org.mockito.MockitoAnnotations;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
public class StreamingJobUtilsTest {
@@ -237,4 +241,13 @@ public class StreamingJobUtilsTest {
Assert.assertNotNull(normalVarcharColumn);
Assert.assertEquals(150, normalVarcharColumn.getType().getLength());
// 50 * 3
}
+
+ @Test
+ public void testGetOceanBaseRemoteDbName() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put(DataSourceConfigKeys.DATABASE, "test_db");
+
+ Assert.assertEquals("test_db",
+ StreamingJobUtils.getRemoteDbName(DataSourceType.OCEANBASE,
properties));
+ }
}
diff --git
a/fs_brokers/cdc_client/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java
b/fs_brokers/cdc_client/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java
index bbf973641e4..275ec709211 100644
---
a/fs_brokers/cdc_client/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java
+++
b/fs_brokers/cdc_client/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java
@@ -84,13 +84,18 @@ import java.util.function.Predicate;
import static io.debezium.util.Strings.isNullOrEmpty;
/**
- * Copied from FlinkCDC project(3.5.0).
+ * Copied from FlinkCDC project(3.6.0).
*
- * <p>Line 924 : change Log Level info to debug.
+ * <p>Line 940 : change Log Level info to debug.
+ *
+ * <p>Line 420 : exclude OceanBase heartbeat events from restart event
counting.
*/
public class MySqlStreamingChangeEventSource
implements StreamingChangeEventSource<MySqlPartition,
MySqlOffsetContext> {
+ public static final String EXCLUDE_HEARTBEAT_FROM_EVENT_COUNT =
+ "doris.cdc.exclude.heartbeat.from.event.count";
+
private static final Logger LOGGER =
LoggerFactory.getLogger(MySqlStreamingChangeEventSource.class);
@@ -103,6 +108,7 @@ public class MySqlStreamingChangeEventSource
private final Clock clock;
private final EventProcessingFailureHandlingMode
eventDeserializationFailureHandlingMode;
private final EventProcessingFailureHandlingMode
inconsistentSchemaHandlingMode;
+ private final boolean excludeHeartbeatFromEventCount;
private int startingRowNumber = 0;
private long initialEventsToSkip = 0L;
@@ -226,6 +232,8 @@ public class MySqlStreamingChangeEventSource
}
}
Configuration configuration = connectorConfig.getConfig();
+ excludeHeartbeatFromEventCount =
+ configuration.getBoolean(EXCLUDE_HEARTBEAT_FROM_EVENT_COUNT,
false);
client.setKeepAlive(configuration.getBoolean(MySqlConnectorConfig.KEEP_ALIVE));
final long keepAliveInterval =
configuration.getLong(MySqlConnectorConfig.KEEP_ALIVE_INTERVAL_MS);
@@ -408,7 +416,10 @@ public class MySqlStreamingChangeEventSource
eventDispatcher.dispatchHeartbeatEvent(partition, offsetContext);
// Capture that we've completed another event ...
- offsetContext.completeEvent();
+ // OceanBase heartbeat events must not participate in restart
event counting.
+ if (shouldCompleteEvent(excludeHeartbeatFromEventCount,
eventType)) {
+ offsetContext.completeEvent();
+ }
// update last offset used for logging
lastOffset = offsetContext.getOffset();
@@ -521,6 +532,11 @@ public class MySqlStreamingChangeEventSource
}
}
+ static boolean shouldCompleteEvent(
+ boolean excludeHeartbeatFromEventCount, EventType eventType) {
+ return !excludeHeartbeatFromEventCount || eventType !=
EventType.HEARTBEAT;
+ }
+
/**
* Handle the supplied event with a {@link RotateEventData} that signals
the logs are being
* rotated. This means that either the server was restarted, or the binlog
has transitioned to a
@@ -937,11 +953,18 @@ public class MySqlStreamingChangeEventSource
int count = 0;
int numRows = rows.size();
if (startingRowNumber < numRows) {
- for (int row = startingRowNumber; row != numRows; ++row) {
- offsetContext.setRowNumber(row, numRows);
- offsetContext.event(tableId, eventTimestamp);
- changeEmitter.emit(tableId, rows.get(row));
- count++;
+ // Use iterator to avoid O(n²) complexity when rows is a
LinkedList
+ // (mysql-binlog-connector-java uses LinkedList in
WriteRowsEventDataDeserializer
+ // and DeleteRowsEventDataDeserializer)
+ int rowIndex = 0;
+ for (U rowData : rows) {
+ if (rowIndex >= startingRowNumber) {
+ offsetContext.setRowNumber(rowIndex, numRows);
+ offsetContext.event(tableId, eventTimestamp);
+ changeEmitter.emit(tableId, rowData);
+ count++;
+ }
+ rowIndex++;
}
if (LOGGER.isDebugEnabled()) {
if (startingRowNumber != 0) {
@@ -1273,8 +1296,8 @@ public class MySqlStreamingChangeEventSource
keyManagers = kmf.getKeyManagers();
} catch (KeyStoreException
- | NoSuchAlgorithmException
- | UnrecoverableKeyException e) {
+ | NoSuchAlgorithmException
+ | UnrecoverableKeyException e) {
throw new DebeziumException("Could not load keystore", e);
}
}
@@ -1288,23 +1311,23 @@ public class MySqlStreamingChangeEventSource
if (ks == null && (sslMode == SSLMode.PREFERRED || sslMode ==
SSLMode.REQUIRED)) {
trustManagers =
new TrustManager[] {
- new X509TrustManager() {
-
- @Override
- public void checkClientTrusted(
- X509Certificate[]
x509Certificates, String s)
- throws CertificateException {}
-
- @Override
- public void checkServerTrusted(
- X509Certificate[]
x509Certificates, String s)
- throws CertificateException {}
-
- @Override
- public X509Certificate[]
getAcceptedIssuers() {
- return new X509Certificate[0];
- }
+ new X509TrustManager() {
+
+ @Override
+ public void checkClientTrusted(
+ X509Certificate[]
x509Certificates, String s)
+ throws CertificateException {}
+
+ @Override
+ public void checkServerTrusted(
+ X509Certificate[]
x509Certificates, String s)
+ throws CertificateException {}
+
+ @Override
+ public X509Certificate[]
getAcceptedIssuers() {
+ return new X509Certificate[0];
}
+ }
};
} else {
TrustManagerFactory tmf =
@@ -1402,7 +1425,6 @@ public class MySqlStreamingChangeEventSource
GtidSet mergedGtidSet;
if (connectorConfig.gtidNewChannelPosition() ==
GtidNewChannelPosition.EARLIEST) {
- final GtidSet knownGtidSet = filteredGtidSet;
LOGGER.info("Using first available positions for new GTID
channels");
final GtidSet relevantAvailableServerGtidSet =
(gtidSourceFilter != null)
@@ -1422,14 +1444,16 @@ public class MySqlStreamingChangeEventSource
// recorded offset in the checkpoint, and the available GTID for
other MySQL instances
// should be completed.
mergedGtidSet =
- GtidUtils.fixRestoredGtidSet(
- GtidUtils.mergeGtidSetInto(
- relevantAvailableServerGtidSet.retainAll(
- uuid ->
knownGtidSet.forServerWithId(uuid) != null),
- purgedServerGtid),
- filteredGtidSet);
+ GtidUtils.fixOldChannelsGtidSet(
+ relevantAvailableServerGtidSet, purgedServerGtid,
filteredGtidSet);
} else {
- mergedGtidSet = availableServerGtidSet.with(filteredGtidSet);
+ LOGGER.info("Using latest positions for new GTID channels");
+ mergedGtidSet =
+ GtidUtils.computeLatestModeGtidSet(
+ availableServerGtidSet,
+ purgedServerGtid,
+ filteredGtidSet,
+ gtidSourceFilter);
}
LOGGER.info("Final merged GTID set to use when connecting to MySQL:
{}", mergedGtidSet);
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/factory/DataSource.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/factory/DataSource.java
index 904cc32a537..fe819b8a5e7 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/factory/DataSource.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/factory/DataSource.java
@@ -19,5 +19,6 @@ package org.apache.doris.cdcclient.source.factory;
public enum DataSource {
MYSQL,
- POSTGRES
+ POSTGRES,
+ OCEANBASE
}
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/factory/SourceReaderFactory.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/factory/SourceReaderFactory.java
index 216c2514f42..c0eb9b35788 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/factory/SourceReaderFactory.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/factory/SourceReaderFactory.java
@@ -19,6 +19,7 @@ package org.apache.doris.cdcclient.source.factory;
import org.apache.doris.cdcclient.source.reader.SourceReader;
import org.apache.doris.cdcclient.source.reader.mysql.MySqlSourceReader;
+import
org.apache.doris.cdcclient.source.reader.oceanbase.OceanBaseSourceReader;
import org.apache.doris.cdcclient.source.reader.postgres.PostgresSourceReader;
import java.util.Map;
@@ -38,6 +39,7 @@ public final class SourceReaderFactory {
static {
register(DataSource.MYSQL, MySqlSourceReader::new);
register(DataSource.POSTGRES, PostgresSourceReader::new);
+ register(DataSource.OCEANBASE, OceanBaseSourceReader::new);
}
private SourceReaderFactory() {}
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
index b9277075513..11075ea2d81 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
@@ -92,6 +92,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import static
io.debezium.connector.mysql.MySqlStreamingChangeEventSource.EXCLUDE_HEARTBEAT_FROM_EVENT_COUNT;
import static
org.apache.doris.cdcclient.common.Constants.DEBEZIUM_HEARTBEAT_INTERVAL_MS;
import static org.apache.doris.cdcclient.utils.ConfigUtil.is13Timestamp;
import static org.apache.doris.cdcclient.utils.ConfigUtil.isJson;
@@ -141,6 +142,11 @@ public class MySqlSourceReader extends
AbstractCdcSourceReader {
this.snapshotReaderContexts = new CopyOnWriteArrayList<>();
}
+ /** Whether server heartbeat events should be excluded from the restart
event count. */
+ protected boolean excludeHeartbeatFromEventCount() {
+ return false;
+ }
+
@Override
public void initialize(String jobId, DataSource dataSource, Map<String,
String> config) {
this.serializer.init(config);
@@ -995,6 +1001,9 @@ public class MySqlSourceReader extends
AbstractCdcSourceReader {
dbzProps.setProperty(
MySqlConnectorConfig.KEEP_ALIVE_INTERVAL_MS.name(),
DEBEZIUM_HEARTBEAT_INTERVAL_MS + "");
+ dbzProps.setProperty(
+ EXCLUDE_HEARTBEAT_FROM_EVENT_COUNT,
+ Boolean.toString(excludeHeartbeatFromEventCount()));
if (cdcConfig.containsKey(DataSourceConfigKeys.SSL_MODE)) {
String normalized =
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/common/DataSourceType.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/oceanbase/OceanBaseSourceReader.java
similarity index 62%
copy from
fe/fe-core/src/main/java/org/apache/doris/job/common/DataSourceType.java
copy to
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/oceanbase/OceanBaseSourceReader.java
index b188f265957..c6509fc1405 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/common/DataSourceType.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/oceanbase/OceanBaseSourceReader.java
@@ -15,9 +15,16 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.job.common;
+package org.apache.doris.cdcclient.source.reader.oceanbase;
-public enum DataSourceType {
- MYSQL,
- POSTGRES
+import org.apache.doris.cdcclient.source.reader.mysql.MySqlSourceReader;
+
+public class OceanBaseSourceReader extends MySqlSourceReader {
+
+ @Override
+ protected boolean excludeHeartbeatFromEventCount() {
+ // OceanBase Binlog Service heartbeat events are not guaranteed to
replay at the same
+ // position, so they must not participate in transaction restart event
counting.
+ return true;
+ }
}
diff --git
a/fs_brokers/cdc_client/src/test/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSourceTest.java
b/fs_brokers/cdc_client/src/test/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSourceTest.java
new file mode 100644
index 00000000000..2f012526b31
--- /dev/null
+++
b/fs_brokers/cdc_client/src/test/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSourceTest.java
@@ -0,0 +1,48 @@
+// 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 io.debezium.connector.mysql;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.github.shyiko.mysql.binlog.event.EventType;
+import org.junit.jupiter.api.Test;
+
+class MySqlStreamingChangeEventSourceTest {
+
+ @Test
+ void oceanBaseHeartbeatDoesNotCompleteEvent() {
+ assertFalse(
+ MySqlStreamingChangeEventSource.shouldCompleteEvent(
+ true, EventType.HEARTBEAT));
+ }
+
+ @Test
+ void oceanBaseBinlogEventCompletesEvent() {
+ assertTrue(
+ MySqlStreamingChangeEventSource.shouldCompleteEvent(
+ true, EventType.WRITE_ROWS));
+ }
+
+ @Test
+ void mysqlHeartbeatKeepsExistingCompletionBehavior() {
+ assertTrue(
+ MySqlStreamingChangeEventSource.shouldCompleteEvent(
+ false, EventType.HEARTBEAT));
+ }
+}
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/CdcClientWriteHarness.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/CdcClientWriteHarness.java
index cdc8dc4a62e..07e5b40a811 100644
---
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/CdcClientWriteHarness.java
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/CdcClientWriteHarness.java
@@ -94,6 +94,57 @@ final class CdcClientWriteHarness implements AutoCloseable {
String offset,
String targetDb,
MockDorisServer mock) {
+ return mysqlCompatible(
+ jobId,
+ "MYSQL",
+ host,
+ port,
+ user,
+ password,
+ database,
+ includeTables,
+ offset,
+ targetDb,
+ mock);
+ }
+
+ static CdcClientWriteHarness oceanbase(
+ String jobId,
+ String host,
+ int port,
+ String user,
+ String password,
+ String database,
+ String includeTables,
+ String offset,
+ String targetDb,
+ MockDorisServer mock) {
+ return mysqlCompatible(
+ jobId,
+ "OCEANBASE",
+ host,
+ port,
+ user,
+ password,
+ database,
+ includeTables,
+ offset,
+ targetDb,
+ mock);
+ }
+
+ private static CdcClientWriteHarness mysqlCompatible(
+ String jobId,
+ String dataSource,
+ String host,
+ int port,
+ String user,
+ String password,
+ String database,
+ String includeTables,
+ String offset,
+ String targetDb,
+ MockDorisServer mock) {
Map<String, String> config = new HashMap<>();
config.put(
DataSourceConfigKeys.JDBC_URL,
@@ -108,7 +159,7 @@ final class CdcClientWriteHarness implements AutoCloseable {
// Point cdc_client's stream-load at the mock BE.
Env.getCurrentEnv().setBackendHttpPort(mock.port());
Env.getCurrentEnv().setClusterToken("test");
- return new CdcClientWriteHarness(jobId, "MYSQL", config, targetDb,
mock);
+ return new CdcClientWriteHarness(jobId, dataSource, config, targetDb,
mock);
}
static CdcClientWriteHarness postgres(
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/OceanBaseSchemaChangeITCase.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/OceanBaseSchemaChangeITCase.java
new file mode 100644
index 00000000000..1de7e36cfb0
--- /dev/null
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/OceanBaseSchemaChangeITCase.java
@@ -0,0 +1,121 @@
+// 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.doris.cdcclient.itcase;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import org.apache.doris.cdcclient.common.Env;
+import org.apache.doris.job.cdc.split.SnapshotSplit;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Connection;
+import java.sql.Statement;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+class OceanBaseSchemaChangeITCase extends OceanBaseTestBase {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+ private static final AtomicLong JOB_ID_SEQ = new AtomicLong(1_300_000);
+
+ private String jobId;
+ private String database;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ jobId = String.valueOf(JOB_ID_SEQ.incrementAndGet());
+ database = "oceanbase_schema_" + jobId;
+ createDatabase(
+ database,
+ "CREATE TABLE t_user (id INT NOT NULL, name VARCHAR(50),
PRIMARY KEY (id))",
+ "INSERT INTO t_user VALUES (1,'alice'), (2,'bob')");
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ Env.getCurrentEnv().close(jobId);
+ dropDatabase(database);
+ }
+
+ @Test
+ void addAndDropColumnSurviveOffsetCommitFailureAndReaderRebuild() throws
Exception {
+ try (MockDorisServer mock = new MockDorisServer();
+ CdcClientWriteHarness harness =
+ oceanBaseHarness(jobId, database, "t_user", "initial",
mock)) {
+ List<SnapshotSplit> splits =
harness.fetchAllSnapshotSplits("t_user");
+ harness.writeSnapshot(splits);
+ harness.enterBinlog(splits);
+ String offsetBeforeAlter = harness.committedOffset();
+ String schemasBeforeAlter = harness.committedTableSchemas();
+ assertThat(schemasBeforeAlter).isNotNull().doesNotContain("city");
+
+ execute(
+ "ALTER TABLE t_user ADD COLUMN city VARCHAR(50)",
+ "INSERT INTO t_user (id, name, city) VALUES (3, 'carol',
'beijing')");
+
+ mock.failCommitOffset();
+ assertThatThrownBy(() -> harness.continueBinlog(1,
Duration.ofSeconds(90)))
+ .isInstanceOf(Exception.class);
+ assertThat(mock.failedCommitOffsetCount()).isPositive();
+ assertThat(harness.committedOffset()).isEqualTo(offsetBeforeAlter);
+
assertThat(harness.committedTableSchemas()).isEqualTo(schemasBeforeAlter);
+
+ mock.allowCommitOffset();
+ harness.rebuildReaderOnNextWrite();
+ List<String> retried = harness.continueBinlog(1,
Duration.ofSeconds(90));
+
+ assertThat(harness.executedDdls()).hasSize(2);
+ assertThat(harness.executedDdls()).allMatch(ddl ->
ddl.contains("ADD COLUMN"));
+ assertThat(mock.ddlResponses().get(1))
+ .contains("Can not add column which already exists");
+ assertThat(retried).hasSize(1);
+ JsonNode added = MAPPER.readTree(retried.get(0));
+ assertThat(added.get("id").asInt()).isEqualTo(3);
+ assertThat(added.get("city").asText()).isEqualTo("beijing");
+ assertThat(harness.committedTableSchemas()).contains("city");
+
+ execute(
+ "ALTER TABLE t_user DROP COLUMN city",
+ "INSERT INTO t_user (id, name) VALUES (4, 'dave')");
+ List<String> afterDrop = harness.continueBinlog(1,
Duration.ofSeconds(90));
+
+ assertThat(harness.executedDdls().get(2)).contains("DROP
COLUMN").contains("city");
+ assertThat(afterDrop).hasSize(1);
+ JsonNode dropped = MAPPER.readTree(afterDrop.get(0));
+ assertThat(dropped.get("id").asInt()).isEqualTo(4);
+ assertThat(dropped.has("city")).isFalse();
+ assertThat(harness.committedTableSchemas()).doesNotContain("city");
+ }
+ }
+
+ private void execute(String... statements) throws Exception {
+ try (Connection connection = connection(database);
+ Statement statement = connection.createStatement()) {
+ for (String sql : statements) {
+ statement.execute(sql);
+ }
+ }
+ }
+}
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/OceanBaseStartupOffsetITCase.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/OceanBaseStartupOffsetITCase.java
new file mode 100644
index 00000000000..7b6980cd9ad
--- /dev/null
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/OceanBaseStartupOffsetITCase.java
@@ -0,0 +1,148 @@
+// 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.doris.cdcclient.itcase;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.apache.doris.cdcclient.common.Env;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+class OceanBaseStartupOffsetITCase extends OceanBaseTestBase {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+ private static final AtomicLong JOB_ID_SEQ = new AtomicLong(1_200_000);
+
+ private String jobId;
+ private String database;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ jobId = String.valueOf(JOB_ID_SEQ.incrementAndGet());
+ database = "oceanbase_offset_" + jobId;
+ createDatabase(
+ database,
+ "CREATE TABLE t_user (id INT NOT NULL, name VARCHAR(50),
PRIMARY KEY (id))",
+ "INSERT INTO t_user VALUES (1,'alice'), (2,'bob')");
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ Env.getCurrentEnv().close(jobId);
+ dropDatabase(database);
+ }
+
+ @Test
+ void latestReadsOnlyChangesAfterReaderStarts() throws Exception {
+ awaitInitialRowsInBinlog();
+
+ try (MockDorisServer mock = new MockDorisServer();
+ CdcClientWriteHarness harness =
+ oceanBaseHarness(jobId, database, "t_user", "latest",
mock)) {
+ harness.enterBinlogFromStartupMode();
+ insert("INSERT INTO t_user VALUES (3,'carol')");
+
+ assertThat(ids(harness.continueBinlog(1, Duration.ofSeconds(90))))
+ .containsExactly(3);
+ assertThat(ids(harness.loadedRecords())).doesNotContain(1, 2);
+ }
+ }
+
+ @Test
+ void earliestReplaysExistingBinlogChanges() throws Exception {
+ try (MockDorisServer mock = new MockDorisServer();
+ CdcClientWriteHarness harness =
+ oceanBaseHarness(jobId, database, "t_user",
"earliest", mock)) {
+ assertThat(ids(harness.readBinlogFromStartupMode(2,
Duration.ofSeconds(90))))
+ .containsExactlyInAnyOrder(1, 2);
+ }
+ }
+
+ @Test
+ void specificOffsetReplaysChangesAfterRecordedPosition() throws Exception {
+ awaitInitialRowsInBinlog();
+
+ String[] position = currentBinlogPosition();
+ String offset =
+ String.format(
+ "{\"file\":\"%s\",\"pos\":\"%s\"}", position[0],
position[1]);
+ insert("INSERT INTO t_user VALUES (3,'carol')");
+
+ try (MockDorisServer mock = new MockDorisServer();
+ CdcClientWriteHarness harness =
+ oceanBaseHarness(jobId, database, "t_user", offset,
mock)) {
+ assertThat(ids(harness.readBinlogFromStartupMode(1,
Duration.ofSeconds(90))))
+ .containsExactly(3);
+ assertThat(ids(harness.loadedRecords())).doesNotContain(1, 2);
+ }
+ }
+
+ private void awaitInitialRowsInBinlog() throws Exception {
+ String probeJobId = String.valueOf(JOB_ID_SEQ.incrementAndGet());
+
+ try (MockDorisServer mock = new MockDorisServer();
+ CdcClientWriteHarness harness =
+ oceanBaseHarness(
+ probeJobId, database, "t_user", "earliest",
mock)) {
+ assertThat(
+ ids(
+ harness.readBinlogFromStartupMode(
+ 2, Duration.ofSeconds(90))))
+ .containsExactlyInAnyOrder(1, 2);
+ }
+ }
+
+ private String[] currentBinlogPosition() throws Exception {
+ try (Connection connection = connection(database);
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery("SHOW MASTER
STATUS")) {
+ if (!resultSet.next()) {
+ throw new IllegalStateException("SHOW MASTER STATUS returned
no row");
+ }
+ return new String[] {resultSet.getString("File"),
resultSet.getString("Position")};
+ }
+ }
+
+ private void insert(String sql) throws Exception {
+ try (Connection connection = connection(database);
+ Statement statement = connection.createStatement()) {
+ statement.execute(sql);
+ }
+ }
+
+ private List<Integer> ids(List<String> records) throws Exception {
+ List<Integer> result = new ArrayList<>();
+ for (String record : records) {
+ JsonNode node = MAPPER.readTree(record);
+ result.add(node.get("id").asInt());
+ }
+ return result;
+ }
+}
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/OceanBaseTestBase.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/OceanBaseTestBase.java
new file mode 100644
index 00000000000..a6157335c3d
--- /dev/null
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/OceanBaseTestBase.java
@@ -0,0 +1,150 @@
+// 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.doris.cdcclient.itcase;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy;
+import org.testcontainers.utility.DockerImageName;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.time.Duration;
+
+abstract class OceanBaseTestBase {
+
+ protected static final int OCEANBASE_CDC_PORT = 2883;
+ protected static final String USER = "root@test";
+ protected static final String PASSWORD = "123456";
+
+ private static final String EXTERNAL_HOST =
System.getProperty("oceanbase.host");
+ private static final int EXTERNAL_CDC_PORT =
+ Integer.getInteger("oceanbase.cdc.port", OCEANBASE_CDC_PORT);
+
+ static final GenericContainer<?> OCEANBASE =
+ new GenericContainer<>(
+
DockerImageName.parse("quay.io/oceanbase/obbinlog-ce:4.2.5-test"))
+ .withExposedPorts(2881, OCEANBASE_CDC_PORT)
+ .withStartupTimeout(Duration.ofMinutes(6))
+ .waitingFor(
+ new LogMessageWaitStrategy()
+ .withRegEx(".*OBBinlog is ready!.*")
+ .withTimes(1)
+
.withStartupTimeout(Duration.ofMinutes(6)));
+
+ @BeforeAll
+ static void initializeOceanBase() {
+ if (!useExternalOceanBase()) {
+ OCEANBASE.start();
+ }
+
+ Awaitility.await()
+ .atMost(60, SECONDS)
+ .pollInterval(1, SECONDS)
+ .ignoreExceptions()
+ .until(
+ () -> {
+ try (Connection connection = connection("");
+ Statement statement =
connection.createStatement();
+ ResultSet resultSet =
+ statement.executeQuery("SHOW
MASTER STATUS")) {
+ return connection.isValid(1)
+ && resultSet.next()
+ &&
!resultSet.getString("File").isEmpty()
+ && resultSet.getLong("Position") >= 4;
+ }
+ });
+ }
+
+ @AfterAll
+ static void stopOceanBase() {
+ if (!useExternalOceanBase()) {
+ OCEANBASE.stop();
+ }
+ }
+
+ protected static Connection connection(String database) throws Exception {
+ String url =
+ "jdbc:mysql://"
+ + oceanBaseHost()
+ + ":"
+ + oceanBaseCdcPort()
+ + "/"
+ + database
+ + "?serverTimezone=UTC";
+ return DriverManager.getConnection(url, USER, PASSWORD);
+ }
+
+ protected void createDatabase(String database, String... statements)
throws Exception {
+ try (Connection connection = connection("");
+ Statement statement = connection.createStatement()) {
+ statement.execute("DROP DATABASE IF EXISTS " + database);
+ statement.execute("CREATE DATABASE " + database);
+ statement.execute("USE " + database);
+ for (String sql : statements) {
+ statement.execute(sql);
+ }
+ }
+ }
+
+ protected void dropDatabase(String database) throws Exception {
+ try (Connection connection = connection("");
+ Statement statement = connection.createStatement()) {
+ statement.execute("DROP DATABASE IF EXISTS " + database);
+ }
+ }
+
+ protected CdcClientWriteHarness oceanBaseHarness(
+ String jobId,
+ String database,
+ String includeTables,
+ String offset,
+ MockDorisServer mock) {
+ return CdcClientWriteHarness.oceanbase(
+ jobId,
+ oceanBaseHost(),
+ oceanBaseCdcPort(),
+ USER,
+ PASSWORD,
+ database,
+ includeTables,
+ offset,
+ "doris_target_db",
+ mock);
+ }
+
+ private static boolean useExternalOceanBase() {
+ return EXTERNAL_HOST != null;
+ }
+
+ private static String oceanBaseHost() {
+ return useExternalOceanBase() ? EXTERNAL_HOST : OCEANBASE.getHost();
+ }
+
+ private static int oceanBaseCdcPort() {
+ return useExternalOceanBase()
+ ? EXTERNAL_CDC_PORT
+ : OCEANBASE.getMappedPort(OCEANBASE_CDC_PORT);
+ }
+}
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/OceanBaseWriteDmlITCase.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/OceanBaseWriteDmlITCase.java
new file mode 100644
index 00000000000..eb7159ab7e7
--- /dev/null
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/OceanBaseWriteDmlITCase.java
@@ -0,0 +1,112 @@
+// 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.doris.cdcclient.itcase;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.apache.doris.cdcclient.common.Constants;
+import org.apache.doris.cdcclient.common.Env;
+import org.apache.doris.job.cdc.split.SnapshotSplit;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Connection;
+import java.sql.Statement;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+class OceanBaseWriteDmlITCase extends OceanBaseTestBase {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+ private static final AtomicLong JOB_ID_SEQ = new AtomicLong(1_100_000);
+
+ private String jobId;
+ private String database;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ jobId = String.valueOf(JOB_ID_SEQ.incrementAndGet());
+ database = "oceanbase_dml_" + jobId;
+ createDatabase(
+ database,
+ "CREATE TABLE t_user (id INT NOT NULL, name VARCHAR(50),
PRIMARY KEY (id))",
+ "INSERT INTO t_user VALUES (1,'alice'), (2,'bob'),
(3,'carol')");
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ Env.getCurrentEnv().close(jobId);
+ dropDatabase(database);
+ }
+
+ @Test
+ void snapshotAndBinlogDmlUseOceanBaseReader() throws Exception {
+ try (MockDorisServer mock = new MockDorisServer();
+ CdcClientWriteHarness harness =
+ oceanBaseHarness(jobId, database, "t_user", "initial",
mock)) {
+ List<SnapshotSplit> splits =
harness.fetchAllSnapshotSplits("t_user");
+ harness.writeSnapshot(splits);
+
+ Map<Integer, JsonNode> snapshotById =
indexById(harness.loadedRecords());
+ assertThat(snapshotById).containsOnlyKeys(1, 2, 3);
+ harness.enterBinlog(splits);
+
+ try (Connection connection = connection(database);
+ Statement statement = connection.createStatement()) {
+ statement.execute("INSERT INTO t_user VALUES (4, 'dave')");
+ statement.execute("UPDATE t_user SET name = 'alice2' WHERE id
= 1");
+ statement.execute("DELETE FROM t_user WHERE id = 2");
+ }
+
+ List<String> binlog = harness.continueBinlog(3,
Duration.ofSeconds(90));
+ assertThat(binlog).hasSize(3);
+ Map<Integer, JsonNode> binlogById = indexById(binlog);
+ assertThat(binlogById).containsOnlyKeys(1, 2, 4);
+
assertThat(binlogById.get(4).get("name").asText()).isEqualTo("dave");
+
assertThat(binlogById.get(4).get(Constants.DORIS_DELETE_SIGN).asInt()).isZero();
+
assertThat(binlogById.get(1).get("name").asText()).isEqualTo("alice2");
+
assertThat(binlogById.get(1).get(Constants.DORIS_DELETE_SIGN).asInt()).isZero();
+
assertThat(binlogById.get(2).get(Constants.DORIS_DELETE_SIGN).asInt()).isEqualTo(1);
+
+ harness.rebuildReaderOnNextWrite();
+ try (Connection connection = connection(database);
+ Statement statement = connection.createStatement()) {
+ statement.execute("INSERT INTO t_user VALUES (5, 'eve')");
+ }
+
+ List<String> resumed = harness.continueBinlog(1,
Duration.ofSeconds(90));
+ assertThat(indexById(resumed)).containsOnlyKeys(5);
+ }
+ }
+
+ private Map<Integer, JsonNode> indexById(List<String> records) throws
Exception {
+ Map<Integer, JsonNode> result = new HashMap<>();
+ for (String record : records) {
+ JsonNode node = MAPPER.readTree(record);
+ result.put(node.get("id").asInt(), node);
+ }
+ return result;
+ }
+}
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/factory/SourceReaderFactoryTest.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/factory/SourceReaderFactoryTest.java
index bee4d048c62..8dd8651221a 100644
---
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/factory/SourceReaderFactoryTest.java
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/factory/SourceReaderFactoryTest.java
@@ -22,6 +22,7 @@ import static org.junit.jupiter.api.Assertions.assertNotSame;
import org.apache.doris.cdcclient.source.reader.SourceReader;
import org.apache.doris.cdcclient.source.reader.mysql.MySqlSourceReader;
+import
org.apache.doris.cdcclient.source.reader.oceanbase.OceanBaseSourceReader;
import org.apache.doris.cdcclient.source.reader.postgres.PostgresSourceReader;
import org.junit.jupiter.api.Test;
@@ -42,6 +43,14 @@ class SourceReaderFactoryTest {
SourceReaderFactory.createSourceReader(DataSource.POSTGRES));
}
+ @Test
+ void createsOceanBaseReader() {
+ SourceReader reader =
SourceReaderFactory.createSourceReader(DataSource.OCEANBASE);
+
+ assertInstanceOf(OceanBaseSourceReader.class, reader);
+ assertInstanceOf(MySqlSourceReader.class, reader);
+ }
+
@Test
void eachCallReturnsFreshInstance() {
SourceReader first =
SourceReaderFactory.createSourceReader(DataSource.MYSQL);
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReaderTest.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReaderTest.java
index 4136e4e617a..8905e0379e2 100644
---
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReaderTest.java
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReaderTest.java
@@ -24,6 +24,7 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import
org.apache.doris.cdcclient.source.reader.oceanbase.OceanBaseSourceReader;
import org.apache.doris.job.cdc.DataSourceConfigKeys;
import org.apache.doris.job.cdc.request.JobBaseConfig;
@@ -48,6 +49,8 @@ import io.debezium.relational.history.TableChanges;
public class MySqlSourceReaderTest {
private static final String SERVER_UUID =
"24bc7850-2c16-11e6-a073-0242ac110002";
+ private static final String EXCLUDE_HEARTBEAT_FROM_EVENT_COUNT =
+ "doris.cdc.exclude.heartbeat.from.event.count";
@Test
void testNormalizeSslModeMapsAllLegalValues() {
@@ -126,6 +129,26 @@ public class MySqlSourceReaderTest {
assertFalse(config.isIncludeSchemaChanges());
}
+ @Test
+ void mysqlHeartbeatContributesToRestartEventCount() throws Exception {
+ MySqlSourceConfig config =
+ sourceConfig(new MySqlSourceReader(), "initial", Map.of());
+
+ assertFalse(
+ config.getDbzConfiguration()
+ .getBoolean(EXCLUDE_HEARTBEAT_FROM_EVENT_COUNT));
+ }
+
+ @Test
+ void oceanBaseHeartbeatDoesNotContributeToRestartEventCount() throws
Exception {
+ MySqlSourceConfig config =
+ sourceConfig(new OceanBaseSourceReader(), "initial", Map.of());
+
+ assertTrue(
+ config.getDbzConfiguration()
+ .getBoolean(EXCLUDE_HEARTBEAT_FROM_EVENT_COUNT));
+ }
+
@Test
void snapshotSplitContainsOnlyCurrentTableSchema() throws Exception {
TableId tableId = TableId.parse("testdb.orders");
@@ -205,6 +228,12 @@ public class MySqlSourceReaderTest {
private MySqlSourceConfig sourceConfig(String offset, Map<String, String>
overrides)
throws Exception {
+ return sourceConfig(new MySqlSourceReader(), offset, overrides);
+ }
+
+ private MySqlSourceConfig sourceConfig(
+ MySqlSourceReader reader, String offset, Map<String, String>
overrides)
+ throws Exception {
Map<String, String> cfg = new HashMap<>();
cfg.put(DataSourceConfigKeys.JDBC_URL,
"jdbc:mysql://localhost:3306/testdb");
cfg.put(DataSourceConfigKeys.USER, "u");
@@ -217,7 +246,7 @@ public class MySqlSourceReaderTest {
MySqlSourceReader.class.getDeclaredMethod(
"generateMySqlConfig", Map.class, String.class,
int.class);
m.setAccessible(true);
- return (MySqlSourceConfig) m.invoke(new MySqlSourceReader(), cfg,
"job-1", 0);
+ return (MySqlSourceConfig) m.invoke(reader, cfg, "job-1", 0);
}
private static Map<String, Object> snapshotOffset(String tableId, String
splitKey) {
diff --git a/regression-test/conf/regression-conf.groovy
b/regression-test/conf/regression-conf.groovy
index ec01b743813..afbe0b1a5c3 100644
--- a/regression-test/conf/regression-conf.groovy
+++ b/regression-test/conf/regression-conf.groovy
@@ -135,6 +135,7 @@ doris_port=9030
mariadb_10_port=3326
db2_11_port=50000
oceanbase_port=2881
+oceanbase_cdc_port=2883
// hive catalog test config
// To enable hive/paimon test, you need first start hive container.
@@ -335,4 +336,4 @@ hudiMinioSecretKey="minio123"
icebergDlfRestCatalog="'type' = 'iceberg', 'warehouse' =
'new_dlf_iceberg_catalog', 'iceberg.catalog.type' = 'rest', 'iceberg.rest.uri'
= 'http://cn-beijing-vpc.dlf.aliyuncs.com/iceberg',
'iceberg.rest.sigv4-enabled' = 'true', 'iceberg.rest.signing-name' = 'DlfNext',
'iceberg.rest.access-key-id' = 'ak', 'iceberg.rest.secret-access-key' = 'sk',
'iceberg.rest.signing-region' = 'cn-beijing',
'iceberg.rest.vended-credentials-enabled' = 'true', 'io-impl' =
'org.apache.iceberg.rest.DlfFileIO', [...]
// For python UDF test, set the runtime version of python, default: 3.8.10
-// pythonUdfRuntimeVersion = ""
\ No newline at end of file
+// pythonUdfRuntimeVersion = ""
diff --git
a/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job.out
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job.out
new file mode 100644
index 00000000000..26ee9fd67b5
--- /dev/null
+++
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job.out
@@ -0,0 +1,17 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !oceanbase_snapshot_users --
+1 Alice 18
+2 Bob 20
+
+-- !oceanbase_snapshot_orders --
+10 snapshot_order
+
+-- !oceanbase_snapshot_empty --
+
+-- !oceanbase_incremental_users --
+2 Bob 21
+3 Carol 30
+
+-- !oceanbase_incremental_orders --
+10 snapshot_order
+11 incremental_order
diff --git
a/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_all_type.out
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_all_type.out
new file mode 100644
index 00000000000..2ee538b7866
--- /dev/null
+++
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_all_type.out
@@ -0,0 +1,8 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !oceanbase_all_type_snapshot --
+1 200 18446744073709551610 123456789.12345 1 2026-07-10
2026-07-10T10:11:12.123456 2026-07-10T10:11:12.123456 abc
snapshot snapshot text 41514944 {"id": 1, "name": "snapshot"}
A,C NEW 4367734D
+2 \N \N \N \N \N \N \N \N \N
\N \N \N \N \N \N
+
+-- !oceanbase_all_type_incremental --
+1 200 18446744073709551610 1.25000 1 2026-07-10
2026-07-10T10:11:12.123456 2026-07-10T02:11:12.123456 abc updated
snapshot text 41514944 {"id":1,"name":"snapshot"} A,C NEW
4367734D
+3 100 9000000000 -98765.43210 0 2026-07-11
2026-07-11T11:12:13.654321 2026-07-11T03:12:13.654321 xyz
incremental incremental text 2F2B3764
{"id":3,"name":"incremental"} B DONE 71383376
diff --git
a/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_offset.out
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_offset.out
new file mode 100644
index 00000000000..8786bd3dd06
--- /dev/null
+++
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_offset.out
@@ -0,0 +1,11 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !oceanbase_offset_latest --
+2 after_latest
+
+-- !oceanbase_offset_earliest --
+1 earliest_one
+2 earliest_two
+
+-- !oceanbase_offset_specific --
+10 specific_one
+11 specific_two
diff --git
a/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_restart_fe.out
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_restart_fe.out
new file mode 100644
index 00000000000..4f8f344724e
--- /dev/null
+++
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_restart_fe.out
@@ -0,0 +1,5 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !oceanbase_restart_fe --
+1 snapshot
+2 updated_after_restart
+3 after_restart
diff --git
a/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_sc_restart_fe.out
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_sc_restart_fe.out
new file mode 100644
index 00000000000..bbda3141447
--- /dev/null
+++
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_sc_restart_fe.out
@@ -0,0 +1,11 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !oceanbase_sc_after_restart --
+1 snapshot \N
+2 before_restart updated_after_restart
+3 after_restart schema_after_restart
+
+-- !oceanbase_sc_restart_final --
+1 snapshot
+2 before_restart
+3 after_restart
+4 after_drop
diff --git
a/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_schema_change.out
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_schema_change.out
new file mode 100644
index 00000000000..a4bebceb4c7
--- /dev/null
+++
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_schema_change.out
@@ -0,0 +1,9 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !oceanbase_after_add --
+1 snapshot \N
+2 after_add Hangzhou
+
+-- !oceanbase_after_drop --
+1 snapshot
+2 after_add
+3 after_drop
diff --git a/regression-test/pipeline/external/conf/regression-conf.groovy
b/regression-test/pipeline/external/conf/regression-conf.groovy
index 5cc507a9a14..d6679434ecc 100644
--- a/regression-test/pipeline/external/conf/regression-conf.groovy
+++ b/regression-test/pipeline/external/conf/regression-conf.groovy
@@ -190,6 +190,7 @@ oracle_11_port=1521
sqlserver_2022_port=1433
clickhouse_22_port=8123
oceanbase_port=2881
+oceanbase_cdc_port=2883
db2_11_port=50000
// trino-connector catalog test config
diff --git a/regression-test/suites/doc/external/jdbc/oceanbase.md.groovy
b/regression-test/suites/doc/external/jdbc/oceanbase.md.groovy
index c17820fbb39..64d3a1efb8d 100644
--- a/regression-test/suites/doc/external/jdbc/oceanbase.md.groovy
+++ b/regression-test/suites/doc/external/jdbc/oceanbase.md.groovy
@@ -30,7 +30,7 @@ suite("oceanbase.md",
"p2,external,oceanbase,external_docker,external_docker_oce
sql """create catalog if not exists ${catalog_name} properties(
"type"="jdbc",
"user"="root@test",
- "password"="",
+ "password"="123456",
"jdbc_url" =
"jdbc:oceanbase://${externalEnvIp}:${oceanbase_port}/doris_test",
"driver_url" = "${driver_url}",
"driver_class" = "com.oceanbase.jdbc.Driver"
diff --git
a/regression-test/suites/external_table_p2/jdbc/test_oceanbase_jdbc_catalog.groovy
b/regression-test/suites/external_table_p2/jdbc/test_oceanbase_jdbc_catalog.groovy
index 7ebc54cd60b..13aadd908ab 100644
---
a/regression-test/suites/external_table_p2/jdbc/test_oceanbase_jdbc_catalog.groovy
+++
b/regression-test/suites/external_table_p2/jdbc/test_oceanbase_jdbc_catalog.groovy
@@ -32,8 +32,8 @@ suite("test_oceanbase_jdbc_catalog", "p2,external") {
sql """ create catalog if not exists ${catalog_name} properties(
"type"="jdbc",
"user"="root@test",
- "password"="",
- "jdbc_url" =
"jdbc:oceanbase://${externalEnvIp}:${oceanbase_port}/doris_test",
+ "password"="123456",
+ "jdbc_url" =
"jdbc:oceanbase://${externalEnvIp}:${oceanbase_port}/doris_test?serverTimezone=UTC",
"driver_url" = "${driver_url}",
"driver_class" = "com.oceanbase.jdbc.Driver"
);"""
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job.groovy
new file mode 100644
index 00000000000..416c8ddc016
--- /dev/null
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job.groovy
@@ -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.
+
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_streaming_oceanbase_job",
+
"p2,external,oceanbase,external_docker,external_docker_oceanbase,nondatalake") {
+ def jobName = "test_streaming_oceanbase_job"
+ def currentDb = (sql "SELECT DATABASE()")[0][0]
+ def sourceDb = "test_oceanbase_streaming_db"
+ def table1 = "oceanbase_streaming_users"
+ def table2 = "oceanbase_streaming_orders"
+ def emptyTable = "oceanbase_streaming_empty"
+
+ sql """DROP JOB IF EXISTS WHERE jobname='${jobName}'"""
+ sql """DROP TABLE IF EXISTS ${currentDb}.${table1} FORCE"""
+ sql """DROP TABLE IF EXISTS ${currentDb}.${table2} FORCE"""
+ sql """DROP TABLE IF EXISTS ${currentDb}.${emptyTable} FORCE"""
+
+ String enabled = context.config.otherConfigs.get("enableJdbcTest")
+ if (enabled != null && enabled.equalsIgnoreCase("true")) {
+ String oceanbaseCdcPort =
context.config.otherConfigs.get("oceanbase_cdc_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String s3Endpoint = getS3Endpoint()
+ String bucket = getS3BucketName()
+ String driverUrl =
+
"https://${bucket}.${s3Endpoint}/regression/jdbc_driver/mysql-connector-j-8.4.0.jar";
+
+ def dumpJobState = {
+ log.info("jobs: " + sql("""SELECT * FROM jobs("type"="insert")
WHERE Name='${jobName}'"""))
+ log.info("tasks: " + sql("""SELECT * FROM tasks("type"="insert")
WHERE JobName='${jobName}'"""))
+ }
+
+ connect("root@test", "123456",
"jdbc:mysql://${externalEnvIp}:${oceanbaseCdcPort}") {
+ sql """CREATE DATABASE IF NOT EXISTS ${sourceDb}"""
+ sql """DROP TABLE IF EXISTS ${sourceDb}.${table1}"""
+ sql """DROP TABLE IF EXISTS ${sourceDb}.${table2}"""
+ sql """DROP TABLE IF EXISTS ${sourceDb}.${emptyTable}"""
+ sql """CREATE TABLE ${sourceDb}.${table1} (
+ id INT NOT NULL,
+ name VARCHAR(100),
+ age INT,
+ PRIMARY KEY (id)
+ ) ENGINE=InnoDB"""
+ sql """CREATE TABLE ${sourceDb}.${table2} (
+ id INT NOT NULL,
+ description VARCHAR(100),
+ PRIMARY KEY (id)
+ ) ENGINE=InnoDB"""
+ sql """CREATE TABLE ${sourceDb}.${emptyTable} (
+ id INT NOT NULL,
+ value VARCHAR(100),
+ PRIMARY KEY (id)
+ ) ENGINE=InnoDB"""
+ sql """INSERT INTO ${sourceDb}.${table1} VALUES
+ (1, 'Alice', 18),
+ (2, 'Bob', 20)"""
+ sql """INSERT INTO ${sourceDb}.${table2} VALUES (10,
'snapshot_order')"""
+ }
+
+ sql """CREATE JOB ${jobName}
+ ON STREAMING
+ FROM OCEANBASE (
+ "jdbc_url" =
"jdbc:mysql://${externalEnvIp}:${oceanbaseCdcPort}",
+ "driver_url" = "${driverUrl}",
+ "driver_class" = "com.mysql.cj.jdbc.Driver",
+ "user" = "root@test",
+ "password" = "123456",
+ "database" = "${sourceDb}",
+ "include_tables" = "${table1},${table2},${emptyTable}",
+ "offset" = "initial"
+ )
+ TO DATABASE ${currentDb} (
+ "table.create.properties.replication_num" = "1"
+ )"""
+
+ try {
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ def users = sql """SELECT COUNT(*) FROM
${currentDb}.${table1}"""
+ def orders = sql """SELECT COUNT(*) FROM
${currentDb}.${table2}"""
+ def emptyTables = sql """SHOW TABLES FROM ${currentDb} LIKE
'${emptyTable}'"""
+ users[0][0] == 2 && orders[0][0] == 1 && emptyTables.size() ==
1
+ })
+ } catch (Exception ex) {
+ dumpJobState()
+ throw ex
+ }
+
+ order_qt_oceanbase_snapshot_users """
+ SELECT id, name, age FROM ${currentDb}.${table1} ORDER BY id
+ """
+ order_qt_oceanbase_snapshot_orders """
+ SELECT id, description FROM ${currentDb}.${table2} ORDER BY id
+ """
+ order_qt_oceanbase_snapshot_empty """
+ SELECT id, value FROM ${currentDb}.${emptyTable} ORDER BY id
+ """
+
+ connect("root@test", "123456",
"jdbc:mysql://${externalEnvIp}:${oceanbaseCdcPort}") {
+ sql """INSERT INTO ${sourceDb}.${table1} VALUES (3, 'Carol', 30)"""
+ sql """UPDATE ${sourceDb}.${table1} SET age=21 WHERE id=2"""
+ sql """DELETE FROM ${sourceDb}.${table1} WHERE id=1"""
+ sql """INSERT INTO ${sourceDb}.${table2} VALUES (11,
'incremental_order')"""
+ }
+
+ try {
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ def users = sql """SELECT id, name, age FROM
${currentDb}.${table1} ORDER BY id"""
+ def orders = sql """SELECT id, description FROM
${currentDb}.${table2} ORDER BY id"""
+ users == [[2, 'Bob', 21], [3, 'Carol', 30]] &&
+ orders == [[10, 'snapshot_order'], [11,
'incremental_order']]
+ })
+ } catch (Exception ex) {
+ dumpJobState()
+ throw ex
+ }
+
+ order_qt_oceanbase_incremental_users """
+ SELECT id, name, age FROM ${currentDb}.${table1} ORDER BY id
+ """
+ order_qt_oceanbase_incremental_orders """
+ SELECT id, description FROM ${currentDb}.${table2} ORDER BY id
+ """
+
+ def status = sql """SELECT Status FROM jobs("type"="insert") WHERE
Name='${jobName}'"""
+ assert status.size() == 1 && status[0][0] == "RUNNING"
+ sql """DROP JOB IF EXISTS WHERE jobname='${jobName}'"""
+ }
+}
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_all_type.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_all_type.groovy
new file mode 100644
index 00000000000..5d81bb0c2bc
--- /dev/null
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_all_type.groovy
@@ -0,0 +1,152 @@
+// 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.
+
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_streaming_oceanbase_job_all_type",
+
"p2,external,oceanbase,external_docker,external_docker_oceanbase,nondatalake") {
+ def jobName = "test_streaming_oceanbase_job_all_type"
+ def currentDb = (sql "SELECT DATABASE()")[0][0]
+ def sourceDb = "test_oceanbase_streaming_db"
+ def table1 = "oceanbase_streaming_all_type"
+
+ sql """DROP JOB IF EXISTS WHERE jobname='${jobName}'"""
+ sql """DROP TABLE IF EXISTS ${currentDb}.${table1} FORCE"""
+
+ String enabled = context.config.otherConfigs.get("enableJdbcTest")
+ if (enabled != null && enabled.equalsIgnoreCase("true")) {
+ String oceanbaseCdcPort =
context.config.otherConfigs.get("oceanbase_cdc_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String s3Endpoint = getS3Endpoint()
+ String bucket = getS3BucketName()
+ String driverUrl =
+
"https://${bucket}.${s3Endpoint}/regression/jdbc_driver/mysql-connector-j-8.4.0.jar";
+ String sourceUrl =
"jdbc:mysql://${externalEnvIp}:${oceanbaseCdcPort}?serverTimezone=UTC"
+
+ def dumpJobState = {
+ log.info("jobs: " + sql("""SELECT * FROM jobs("type"="insert")
WHERE Name='${jobName}'"""))
+ log.info("tasks: " + sql("""SELECT * FROM tasks("type"="insert")
WHERE JobName='${jobName}'"""))
+ }
+
+ connect("root@test", "123456", sourceUrl) {
+ sql """CREATE DATABASE IF NOT EXISTS ${sourceDb}"""
+ sql """DROP TABLE IF EXISTS ${sourceDb}.${table1}"""
+ sql """CREATE TABLE ${sourceDb}.${table1} (
+ id INT NOT NULL,
+ tiny_unsigned TINYINT UNSIGNED,
+ big_unsigned BIGINT UNSIGNED,
+ amount DECIMAL(18, 5),
+ enabled BOOLEAN,
+ event_date DATE,
+ event_datetime DATETIME(6),
+ event_timestamp TIMESTAMP(6) NULL,
+ fixed_text CHAR(5),
+ variable_text VARCHAR(32),
+ long_text TEXT,
+ binary_value BLOB,
+ json_value JSON,
+ set_value SET('A', 'B', 'C'),
+ enum_value ENUM('NEW', 'DONE'),
+ varbinary_value VARBINARY(16),
+ PRIMARY KEY (id)
+ ) ENGINE=InnoDB"""
+ sql """INSERT INTO ${sourceDb}.${table1} VALUES (
+ 1, 200, 18446744073709551610, 123456789.12345, TRUE,
+ '2026-07-10', '2026-07-10 10:11:12.123456',
+ '2026-07-10 10:11:12.123456', 'abc', 'snapshot',
+ 'snapshot text', X'010203',
'{"id":1,"name":"snapshot"}',
+ 'A,C', 'NEW', X'0A0B0C'
+ )"""
+ sql """INSERT INTO ${sourceDb}.${table1} VALUES (
+ 2, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL,
NULL,
+ NULL, NULL, NULL, NULL, NULL, NULL
+ )"""
+ }
+
+ sql """CREATE JOB ${jobName}
+ ON STREAMING
+ FROM OCEANBASE (
+ "jdbc_url" = "${sourceUrl}",
+ "driver_url" = "${driverUrl}",
+ "driver_class" = "com.mysql.cj.jdbc.Driver",
+ "user" = "root@test",
+ "password" = "123456",
+ "database" = "${sourceDb}",
+ "include_tables" = "${table1}",
+ "offset" = "initial"
+ )
+ TO DATABASE ${currentDb} (
+ "table.create.properties.replication_num" = "1"
+ )"""
+
+ try {
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ def rows = sql """SELECT COUNT(*) FROM
${currentDb}.${table1}"""
+ rows.size() == 1 && rows[0][0] == 2
+ })
+ } catch (Exception ex) {
+ dumpJobState()
+ throw ex
+ }
+
+ order_qt_oceanbase_all_type_snapshot """
+ SELECT id, tiny_unsigned, big_unsigned, amount, enabled,
event_date,
+ event_datetime, event_timestamp, fixed_text, variable_text,
+ long_text, HEX(binary_value), CAST(json_value AS STRING),
+ set_value, enum_value, HEX(varbinary_value)
+ FROM ${currentDb}.${table1}
+ ORDER BY id
+ """
+
+ connect("root@test", "123456", sourceUrl) {
+ sql """INSERT INTO ${sourceDb}.${table1} VALUES (
+ 3, 100, 9000000000, -98765.43210, FALSE,
+ '2026-07-11', '2026-07-11 11:12:13.654321',
+ '2026-07-11 11:12:13.654321', 'xyz', 'incremental',
+ 'incremental text', X'FFEEDD',
'{"id":3,"name":"incremental"}',
+ 'B', 'DONE', X'ABCDEF'
+ )"""
+ sql """UPDATE ${sourceDb}.${table1}
+ SET variable_text='updated', amount=1.25000 WHERE id=1"""
+ sql """DELETE FROM ${sourceDb}.${table1} WHERE id=2"""
+ }
+
+ try {
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ def count = sql """SELECT COUNT(*) FROM
${currentDb}.${table1}"""
+ def updated = sql """SELECT variable_text FROM
${currentDb}.${table1} WHERE id=1"""
+ def inserted = sql """SELECT COUNT(*) FROM
${currentDb}.${table1} WHERE id=3"""
+ count[0][0] == 2 && updated.size() == 1 && updated[0][0] ==
'updated' &&
+ inserted[0][0] == 1
+ })
+ } catch (Exception ex) {
+ dumpJobState()
+ throw ex
+ }
+
+ order_qt_oceanbase_all_type_incremental """
+ SELECT id, tiny_unsigned, big_unsigned, amount, enabled,
event_date,
+ event_datetime, event_timestamp, fixed_text, variable_text,
+ long_text, HEX(binary_value), CAST(json_value AS STRING),
+ set_value, enum_value, HEX(varbinary_value)
+ FROM ${currentDb}.${table1}
+ ORDER BY id
+ """
+
+ sql """DROP JOB IF EXISTS WHERE jobname='${jobName}'"""
+ }
+}
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_offset.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_offset.groovy
new file mode 100644
index 00000000000..3fa947c87c5
--- /dev/null
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_offset.groovy
@@ -0,0 +1,203 @@
+// 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.
+
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_streaming_oceanbase_job_offset",
+
"p2,external,oceanbase,external_docker,external_docker_oceanbase,nondatalake") {
+ def currentDb = (sql "SELECT DATABASE()")[0][0]
+ def sourceDb = "test_oceanbase_streaming_db"
+ def suffix = UUID.randomUUID().toString().replace("-", "").substring(0, 8)
+ def latestJob = "test_streaming_oceanbase_offset_latest"
+ def earliestJob = "test_streaming_oceanbase_offset_earliest"
+ def specificJob = "test_streaming_oceanbase_offset_specific"
+ def latestTable = "oceanbase_offset_latest_${suffix}"
+ def earliestTable = "oceanbase_offset_earliest_${suffix}"
+ def specificTable = "oceanbase_offset_specific_${suffix}"
+
+ [latestJob, earliestJob, specificJob].each {
+ sql """DROP JOB IF EXISTS WHERE jobname='${it}'"""
+ }
+ [latestTable, earliestTable, specificTable].each {
+ sql """DROP TABLE IF EXISTS ${currentDb}.${it} FORCE"""
+ }
+
+ String enabled = context.config.otherConfigs.get("enableJdbcTest")
+ if (enabled != null && enabled.equalsIgnoreCase("true")) {
+ String oceanbaseCdcPort =
context.config.otherConfigs.get("oceanbase_cdc_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String s3Endpoint = getS3Endpoint()
+ String bucket = getS3BucketName()
+ String driverUrl =
+
"https://${bucket}.${s3Endpoint}/regression/jdbc_driver/mysql-connector-j-8.4.0.jar";
+ String sourceUrl = "jdbc:mysql://${externalEnvIp}:${oceanbaseCdcPort}"
+
+ def waitForJobReady = { String jobName ->
+ Awaitility.await().atMost(180, SECONDS).pollInterval(2,
SECONDS).until({
+ def rows = sql """SELECT Status, SucceedTaskCount FROM
jobs("type"="insert")
+ WHERE Name='${jobName}'"""
+ rows.size() == 1 && rows[0][0] == "RUNNING" && (rows[0][1] as
int) >= 1
+ })
+ }
+ def waitForRows = { String tableName, int count ->
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ def rows = sql """SELECT COUNT(*) FROM
${currentDb}.${tableName}"""
+ rows.size() == 1 && rows[0][0] == count
+ })
+ }
+ def waitForRowId = { String tableName, int id ->
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ def rows = sql """SELECT COUNT(*) FROM
${currentDb}.${tableName} WHERE id = ${id}"""
+ rows.size() == 1 && rows[0][0] == 1
+ })
+ }
+ def dumpJobState = { String jobName ->
+ log.info("jobs: " + sql("""SELECT * FROM jobs("type"="insert")
WHERE Name='${jobName}'"""))
+ log.info("tasks: " + sql("""SELECT * FROM tasks("type"="insert")
WHERE JobName='${jobName}'"""))
+ }
+
+ connect("root@test", "123456", sourceUrl) {
+ sql """CREATE DATABASE IF NOT EXISTS ${sourceDb}"""
+ sql """DROP TABLE IF EXISTS ${sourceDb}.${latestTable}"""
+ sql """CREATE TABLE ${sourceDb}.${latestTable} (
+ id INT NOT NULL,
+ name VARCHAR(100),
+ PRIMARY KEY (id)
+ ) ENGINE=InnoDB"""
+ sql """INSERT INTO ${sourceDb}.${latestTable} VALUES (1,
'before_latest')"""
+ }
+
+ // Consume later records first to ensure before_latest is visible in
OBBinlog.
+ def binlogFile = ""
+ def binlogPosition = ""
+ connect("root@test", "123456", sourceUrl) {
+ sql """DROP TABLE IF EXISTS ${sourceDb}.${specificTable}"""
+ sql """CREATE TABLE ${sourceDb}.${specificTable} (
+ id INT NOT NULL,
+ name VARCHAR(100),
+ PRIMARY KEY (id)
+ ) ENGINE=InnoDB"""
+ def masterStatus = sql """SHOW MASTER STATUS"""
+ binlogFile = masterStatus[0][0]
+ binlogPosition = masterStatus[0][1].toString()
+ sql """INSERT INTO ${sourceDb}.${specificTable} VALUES
+ (10, 'specific_one'),
+ (11, 'specific_two')"""
+ }
+ def offsetJson =
"""{"file":"${binlogFile}","pos":"${binlogPosition}"}"""
+
+ sql """CREATE JOB ${specificJob}
+ ON STREAMING
+ FROM OCEANBASE (
+ "jdbc_url" = "${sourceUrl}",
+ "driver_url" = "${driverUrl}",
+ "driver_class" = "com.mysql.cj.jdbc.Driver",
+ "user" = "root@test",
+ "password" = "123456",
+ "database" = "${sourceDb}",
+ "include_tables" = "${specificTable}",
+ "offset" = '${offsetJson}'
+ )
+ TO DATABASE ${currentDb} (
+ "table.create.properties.replication_num" = "1"
+ )"""
+
+ try {
+ waitForRows(specificTable, 2)
+ } catch (Exception ex) {
+ dumpJobState(specificJob)
+ throw ex
+ }
+ sql """DROP JOB IF EXISTS WHERE jobname='${specificJob}'"""
+
+ sql """CREATE JOB ${latestJob}
+ ON STREAMING
+ FROM OCEANBASE (
+ "jdbc_url" = "${sourceUrl}",
+ "driver_url" = "${driverUrl}",
+ "driver_class" = "com.mysql.cj.jdbc.Driver",
+ "user" = "root@test",
+ "password" = "123456",
+ "database" = "${sourceDb}",
+ "include_tables" = "${latestTable}",
+ "offset" = "latest"
+ )
+ TO DATABASE ${currentDb} (
+ "table.create.properties.replication_num" = "1"
+ )"""
+
+ try {
+ waitForJobReady(latestJob)
+ connect("root@test", "123456", sourceUrl) {
+ sql """INSERT INTO ${sourceDb}.${latestTable} VALUES (2,
'after_latest')"""
+ }
+ waitForRowId(latestTable, 2)
+ } catch (Exception ex) {
+ dumpJobState(latestJob)
+ throw ex
+ }
+
+ order_qt_oceanbase_offset_latest """
+ SELECT id, name FROM ${currentDb}.${latestTable} ORDER BY id
+ """
+ sql """DROP JOB IF EXISTS WHERE jobname='${latestJob}'"""
+
+ connect("root@test", "123456", sourceUrl) {
+ sql """DROP TABLE IF EXISTS ${sourceDb}.${earliestTable}"""
+ sql """CREATE TABLE ${sourceDb}.${earliestTable} (
+ id INT NOT NULL,
+ name VARCHAR(100),
+ PRIMARY KEY (id)
+ ) ENGINE=InnoDB"""
+ sql """INSERT INTO ${sourceDb}.${earliestTable} VALUES
+ (1, 'earliest_one'),
+ (2, 'earliest_two')"""
+ }
+
+ sql """CREATE JOB ${earliestJob}
+ ON STREAMING
+ FROM OCEANBASE (
+ "jdbc_url" = "${sourceUrl}",
+ "driver_url" = "${driverUrl}",
+ "driver_class" = "com.mysql.cj.jdbc.Driver",
+ "user" = "root@test",
+ "password" = "123456",
+ "database" = "${sourceDb}",
+ "include_tables" = "${earliestTable}",
+ "offset" = "earliest"
+ )
+ TO DATABASE ${currentDb} (
+ "table.create.properties.replication_num" = "1"
+ )"""
+
+ try {
+ waitForRows(earliestTable, 2)
+ } catch (Exception ex) {
+ dumpJobState(earliestJob)
+ throw ex
+ }
+
+ order_qt_oceanbase_offset_earliest """
+ SELECT id, name FROM ${currentDb}.${earliestTable} ORDER BY id
+ """
+ sql """DROP JOB IF EXISTS WHERE jobname='${earliestJob}'"""
+
+ order_qt_oceanbase_offset_specific """
+ SELECT id, name FROM ${currentDb}.${specificTable} ORDER BY id
+ """
+ }
+}
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_restart_fe.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_restart_fe.groovy
new file mode 100644
index 00000000000..d61732a31f9
--- /dev/null
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_restart_fe.groovy
@@ -0,0 +1,150 @@
+// 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.
+
+import org.apache.doris.regression.suite.ClusterOptions
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_streaming_oceanbase_job_restart_fe",
+
"p2,docker,oceanbase,external_docker,external_docker_oceanbase,nondatalake") {
+ def jobName = "test_streaming_oceanbase_job_restart_fe"
+ def sourceDb = "test_oceanbase_streaming_db"
+ def table1 = "oceanbase_streaming_restart_fe"
+ def options = new ClusterOptions()
+ options.setFeNum(1)
+ options.cloudMode = null
+
+ docker(options) {
+ def currentDb = (sql "SELECT DATABASE()")[0][0]
+
+ sql """DROP JOB IF EXISTS WHERE jobname='${jobName}'"""
+ sql """DROP TABLE IF EXISTS ${currentDb}.${table1} FORCE"""
+
+ String enabled = context.config.otherConfigs.get("enableJdbcTest")
+ if (enabled != null && enabled.equalsIgnoreCase("true")) {
+ String oceanbaseCdcPort =
context.config.otherConfigs.get("oceanbase_cdc_port")
+ String externalEnvIp =
context.config.otherConfigs.get("externalEnvIp")
+ String s3Endpoint = getS3Endpoint()
+ String bucket = getS3BucketName()
+ String driverUrl =
+
"https://${bucket}.${s3Endpoint}/regression/jdbc_driver/mysql-connector-j-8.4.0.jar";
+ String sourceUrl =
"jdbc:mysql://${externalEnvIp}:${oceanbaseCdcPort}"
+
+ def waitForValue = { int id, String expected ->
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ def rows = sql "SELECT value FROM ${currentDb}.${table1}
WHERE id=${id}"
+ rows.size() == 1 && rows[0][0] == expected
+ })
+ }
+ def waitForJobAfterRestart = {
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS)
+ .ignoreExceptions().until({
+ context.reconnectFe()
+ def rows = sql """SELECT Status FROM
jobs("type"="insert")
+ WHERE Name='${jobName}'"""
+ rows.size() == 1 && rows[0][0] == "RUNNING"
+ })
+ }
+ def dumpJobState = {
+ log.info("jobs: " + sql("""SELECT * FROM jobs("type"="insert")
WHERE Name='${jobName}'"""))
+ log.info("tasks: " + sql("""SELECT * FROM
tasks("type"="insert") WHERE JobName='${jobName}'"""))
+ }
+
+ connect("root@test", "123456", sourceUrl) {
+ sql """CREATE DATABASE IF NOT EXISTS ${sourceDb}"""
+ sql """DROP TABLE IF EXISTS ${sourceDb}.${table1}"""
+ sql """CREATE TABLE ${sourceDb}.${table1} (
+ id INT NOT NULL,
+ value VARCHAR(100),
+ PRIMARY KEY (id)
+ ) ENGINE=InnoDB"""
+ sql """INSERT INTO ${sourceDb}.${table1} VALUES (1,
'snapshot')"""
+ }
+
+ sql """CREATE JOB ${jobName}
+ ON STREAMING
+ FROM OCEANBASE (
+ "jdbc_url" = "${sourceUrl}",
+ "driver_url" = "${driverUrl}",
+ "driver_class" = "com.mysql.cj.jdbc.Driver",
+ "user" = "root@test",
+ "password" = "123456",
+ "database" = "${sourceDb}",
+ "include_tables" = "${table1}",
+ "offset" = "initial"
+ )
+ TO DATABASE ${currentDb} (
+ "table.create.properties.replication_num" = "1"
+ )"""
+
+ try {
+ waitForValue(1, "snapshot")
+ connect("root@test", "123456", sourceUrl) {
+ sql """INSERT INTO ${sourceDb}.${table1} VALUES (2,
'before_restart')"""
+ }
+ waitForValue(2, "before_restart")
+ } catch (Exception ex) {
+ dumpJobState()
+ throw ex
+ }
+
+ def beforeOffset = null
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ def rows = sql """SELECT currentOffset FROM
jobs("type"="insert")
+ WHERE Name='${jobName}'"""
+ if (rows.size() != 1 || rows[0][0] == null) {
+ return false
+ }
+ def parsed = parseJson(rows[0][0])
+ if (parsed.file == null || parsed.pos == null) {
+ return false
+ }
+ beforeOffset = parsed
+ return true
+ })
+
+ cluster.restartFrontends()
+ waitForJobAfterRestart()
+ context.reconnectFe()
+
+ def afterOffsetRows = sql """SELECT currentOffset FROM
jobs("type"="insert")
+ WHERE Name='${jobName}'"""
+ assert afterOffsetRows.size() == 1 && afterOffsetRows[0][0] != null
+ def afterOffset = parseJson(afterOffsetRows[0][0])
+ assert afterOffset.file > beforeOffset.file ||
+ (afterOffset.file == beforeOffset.file &&
+ (afterOffset.pos as long) >= (beforeOffset.pos as
long))
+
+ connect("root@test", "123456", sourceUrl) {
+ sql """INSERT INTO ${sourceDb}.${table1} VALUES (3,
'after_restart')"""
+ sql """UPDATE ${sourceDb}.${table1} SET
value='updated_after_restart' WHERE id=2"""
+ }
+
+ try {
+ waitForValue(3, "after_restart")
+ waitForValue(2, "updated_after_restart")
+ } catch (Exception ex) {
+ dumpJobState()
+ throw ex
+ }
+
+ order_qt_oceanbase_restart_fe """
+ SELECT id, value FROM ${currentDb}.${table1} ORDER BY id
+ """
+ sql """DROP JOB IF EXISTS WHERE jobname='${jobName}'"""
+ }
+ }
+}
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_sc_restart_fe.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_sc_restart_fe.groovy
new file mode 100644
index 00000000000..fb07ad59028
--- /dev/null
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_sc_restart_fe.groovy
@@ -0,0 +1,156 @@
+// 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.
+
+import org.apache.doris.regression.suite.ClusterOptions
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_streaming_oceanbase_job_sc_restart_fe",
+
"p2,docker,oceanbase,external_docker,external_docker_oceanbase,nondatalake") {
+ def jobName = "test_streaming_oceanbase_job_sc_restart_fe"
+ def sourceDb = "test_oceanbase_streaming_db"
+ def table1 = "oceanbase_streaming_sc_restart_fe"
+ def options = new ClusterOptions()
+ options.setFeNum(1)
+ options.cloudMode = null
+
+ docker(options) {
+ def currentDb = (sql "SELECT DATABASE()")[0][0]
+
+ sql """DROP JOB IF EXISTS WHERE jobname='${jobName}'"""
+ sql """DROP TABLE IF EXISTS ${currentDb}.${table1} FORCE"""
+
+ String enabled = context.config.otherConfigs.get("enableJdbcTest")
+ if (enabled != null && enabled.equalsIgnoreCase("true")) {
+ String oceanbaseCdcPort =
context.config.otherConfigs.get("oceanbase_cdc_port")
+ String externalEnvIp =
context.config.otherConfigs.get("externalEnvIp")
+ String s3Endpoint = getS3Endpoint()
+ String bucket = getS3BucketName()
+ String driverUrl =
+
"https://${bucket}.${s3Endpoint}/regression/jdbc_driver/mysql-connector-j-8.4.0.jar";
+ String sourceUrl =
"jdbc:mysql://${externalEnvIp}:${oceanbaseCdcPort}"
+
+ def waitForColumn = { String column, boolean expected ->
+ Awaitility.await().atMost(180, SECONDS).pollInterval(2,
SECONDS).until({
+ (sql "DESC ${currentDb}.${table1}").any { it[0] == column
} == expected
+ })
+ }
+ def waitForValue = { int id, String column, String expected ->
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ def rows = sql "SELECT ${column} FROM
${currentDb}.${table1} WHERE id=${id}"
+ rows.size() == 1 && String.valueOf(rows[0][0]) == expected
+ })
+ }
+ def waitForJobAfterRestart = {
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS)
+ .ignoreExceptions().until({
+ context.reconnectFe()
+ def rows = sql """SELECT Status FROM
jobs("type"="insert")
+ WHERE Name='${jobName}'"""
+ rows.size() == 1 && rows[0][0] == "RUNNING"
+ })
+ }
+ def dumpJobState = {
+ log.info("jobs: " + sql("""SELECT * FROM jobs("type"="insert")
WHERE Name='${jobName}'"""))
+ log.info("tasks: " + sql("""SELECT * FROM
tasks("type"="insert") WHERE JobName='${jobName}'"""))
+ }
+
+ connect("root@test", "123456", sourceUrl) {
+ sql """CREATE DATABASE IF NOT EXISTS ${sourceDb}"""
+ sql """DROP TABLE IF EXISTS ${sourceDb}.${table1}"""
+ sql """CREATE TABLE ${sourceDb}.${table1} (
+ id INT NOT NULL,
+ value VARCHAR(100),
+ PRIMARY KEY (id)
+ ) ENGINE=InnoDB"""
+ sql """INSERT INTO ${sourceDb}.${table1} VALUES (1,
'snapshot')"""
+ }
+
+ sql """CREATE JOB ${jobName}
+ ON STREAMING
+ FROM OCEANBASE (
+ "jdbc_url" = "${sourceUrl}",
+ "driver_url" = "${driverUrl}",
+ "driver_class" = "com.mysql.cj.jdbc.Driver",
+ "user" = "root@test",
+ "password" = "123456",
+ "database" = "${sourceDb}",
+ "include_tables" = "${table1}",
+ "offset" = "initial"
+ )
+ TO DATABASE ${currentDb} (
+ "table.create.properties.replication_num" = "1"
+ )"""
+
+ try {
+ waitForValue(1, "value", "snapshot")
+ connect("root@test", "123456", sourceUrl) {
+ sql """ALTER TABLE ${sourceDb}.${table1} ADD COLUMN
extra_value VARCHAR(50)"""
+ sql """INSERT INTO ${sourceDb}.${table1}
+ VALUES (2, 'before_restart',
'schema_before_restart')"""
+ }
+ waitForColumn("extra_value", true)
+ waitForValue(2, "extra_value", "schema_before_restart")
+ } catch (Exception ex) {
+ dumpJobState()
+ throw ex
+ }
+
+ cluster.restartFrontends()
+ waitForJobAfterRestart()
+ context.reconnectFe()
+
+ connect("root@test", "123456", sourceUrl) {
+ sql """INSERT INTO ${sourceDb}.${table1}
+ VALUES (3, 'after_restart', 'schema_after_restart')"""
+ sql """UPDATE ${sourceDb}.${table1}
+ SET extra_value='updated_after_restart' WHERE id=2"""
+ }
+
+ try {
+ waitForValue(3, "extra_value", "schema_after_restart")
+ waitForValue(2, "extra_value", "updated_after_restart")
+ } catch (Exception ex) {
+ dumpJobState()
+ throw ex
+ }
+
+ order_qt_oceanbase_sc_after_restart """
+ SELECT id, value, extra_value FROM ${currentDb}.${table1}
ORDER BY id
+ """
+
+ connect("root@test", "123456", sourceUrl) {
+ sql """ALTER TABLE ${sourceDb}.${table1} DROP COLUMN
extra_value"""
+ sql """INSERT INTO ${sourceDb}.${table1} VALUES (4,
'after_drop')"""
+ }
+
+ try {
+ waitForColumn("extra_value", false)
+ waitForValue(4, "value", "after_drop")
+ } catch (Exception ex) {
+ dumpJobState()
+ throw ex
+ }
+
+ order_qt_oceanbase_sc_restart_final """
+ SELECT id, value FROM ${currentDb}.${table1} ORDER BY id
+ """
+ def status = sql """SELECT Status FROM jobs("type"="insert") WHERE
Name='${jobName}'"""
+ assert status.size() == 1 && status[0][0] == "RUNNING"
+ sql """DROP JOB IF EXISTS WHERE jobname='${jobName}'"""
+ }
+ }
+}
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_schema_change.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_schema_change.groovy
new file mode 100644
index 00000000000..0535ec60792
--- /dev/null
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_oceanbase_job_schema_change.groovy
@@ -0,0 +1,138 @@
+// 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.
+
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_streaming_oceanbase_job_schema_change",
+
"p2,external,oceanbase,external_docker,external_docker_oceanbase,nondatalake") {
+ def jobName = "test_streaming_oceanbase_job_schema_change"
+ def currentDb = (sql "SELECT DATABASE()")[0][0]
+ def sourceDb = "test_oceanbase_streaming_db"
+ def table1 = "oceanbase_streaming_schema_change"
+
+ sql """DROP JOB IF EXISTS WHERE jobname='${jobName}'"""
+ sql """DROP TABLE IF EXISTS ${currentDb}.${table1} FORCE"""
+
+ String enabled = context.config.otherConfigs.get("enableJdbcTest")
+ if (enabled != null && enabled.equalsIgnoreCase("true")) {
+ String oceanbaseCdcPort =
context.config.otherConfigs.get("oceanbase_cdc_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String s3Endpoint = getS3Endpoint()
+ String bucket = getS3BucketName()
+ String driverUrl =
+
"https://${bucket}.${s3Endpoint}/regression/jdbc_driver/mysql-connector-j-8.4.0.jar";
+
+ def waitForColumn = { String column, boolean expected ->
+ Awaitility.await().atMost(180, SECONDS).pollInterval(2,
SECONDS).until({
+ (sql "DESC ${currentDb}.${table1}").any { it[0] == column } ==
expected
+ })
+ }
+ def waitForValue = { int id, String column, String expected ->
+ Awaitility.await().atMost(180, SECONDS).pollInterval(2,
SECONDS).until({
+ def rows = sql "SELECT ${column} FROM ${currentDb}.${table1}
WHERE id=${id}"
+ rows.size() == 1 && String.valueOf(rows[0][0]) == expected
+ })
+ }
+ def dumpJobState = {
+ log.info("jobs: " + sql("""SELECT * FROM jobs("type"="insert")
WHERE Name='${jobName}'"""))
+ log.info("tasks: " + sql("""SELECT * FROM tasks("type"="insert")
WHERE JobName='${jobName}'"""))
+ }
+
+ connect("root@test", "123456",
"jdbc:mysql://${externalEnvIp}:${oceanbaseCdcPort}") {
+ sql """CREATE DATABASE IF NOT EXISTS ${sourceDb}"""
+ sql """DROP TABLE IF EXISTS ${sourceDb}.${table1}"""
+ sql """CREATE TABLE ${sourceDb}.${table1} (
+ id INT NOT NULL,
+ name VARCHAR(100),
+ PRIMARY KEY (id)
+ ) ENGINE=InnoDB"""
+ sql """INSERT INTO ${sourceDb}.${table1} VALUES (1, 'snapshot')"""
+ }
+
+ sql """CREATE JOB ${jobName}
+ ON STREAMING
+ FROM OCEANBASE (
+ "jdbc_url" =
"jdbc:mysql://${externalEnvIp}:${oceanbaseCdcPort}",
+ "driver_url" = "${driverUrl}",
+ "driver_class" = "com.mysql.cj.jdbc.Driver",
+ "user" = "root@test",
+ "password" = "123456",
+ "database" = "${sourceDb}",
+ "include_tables" = "${table1}",
+ "offset" = "initial"
+ )
+ TO DATABASE ${currentDb} (
+ "table.create.properties.replication_num" = "1"
+ )"""
+
+ try {
+ waitForValue(1, "name", "snapshot")
+ } catch (Exception ex) {
+ dumpJobState()
+ throw ex
+ }
+
+ connect("root@test", "123456",
"jdbc:mysql://${externalEnvIp}:${oceanbaseCdcPort}") {
+ sql """ALTER TABLE ${sourceDb}.${table1} ADD COLUMN city
VARCHAR(50)"""
+ sql """INSERT INTO ${sourceDb}.${table1} VALUES (2, 'after_add',
'Hangzhou')"""
+ }
+
+ try {
+ waitForColumn("city", true)
+ waitForValue(2, "city", "Hangzhou")
+ } catch (Exception ex) {
+ dumpJobState()
+ throw ex
+ }
+
+ order_qt_oceanbase_after_add """
+ SELECT id, name, city FROM ${currentDb}.${table1} ORDER BY id
+ """
+
+ connect("root@test", "123456",
"jdbc:mysql://${externalEnvIp}:${oceanbaseCdcPort}") {
+ sql """UPDATE ${sourceDb}.${table1} SET city='Shanghai' WHERE
id=2"""
+ }
+
+ try {
+ waitForValue(2, "city", "Shanghai")
+ } catch (Exception ex) {
+ dumpJobState()
+ throw ex
+ }
+
+ connect("root@test", "123456",
"jdbc:mysql://${externalEnvIp}:${oceanbaseCdcPort}") {
+ sql """ALTER TABLE ${sourceDb}.${table1} DROP COLUMN city"""
+ sql """INSERT INTO ${sourceDb}.${table1} VALUES (3,
'after_drop')"""
+ }
+
+ try {
+ waitForColumn("city", false)
+ waitForValue(3, "name", "after_drop")
+ } catch (Exception ex) {
+ dumpJobState()
+ throw ex
+ }
+
+ order_qt_oceanbase_after_drop """
+ SELECT id, name FROM ${currentDb}.${table1} ORDER BY id
+ """
+
+ def status = sql """SELECT Status FROM jobs("type"="insert") WHERE
Name='${jobName}'"""
+ assert status.size() == 1 && status[0][0] == "RUNNING"
+ sql """DROP JOB IF EXISTS WHERE jobname='${jobName}'"""
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
