This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 52bbd0a258 [Feature][Connector-V2][MySQL-CDC] Support GTID specific
startup offset (#11123)
52bbd0a258 is described below
commit 52bbd0a2584000d036d017da8d9abd6f22360cc0
Author: Goutam Adwant <[email protected]>
AuthorDate: Tue Jul 21 17:01:51 2026 +0200
[Feature][Connector-V2][MySQL-CDC] Support GTID specific startup offset
(#11123)
Signed-off-by: goutamadwant <[email protected]>
---
docs/en/connectors/source/MySQL-CDC.md | 3 +
docs/zh/connectors/source/MySQL-CDC.md | 3 +
.../connectors/cdc/base/config/StartupConfig.java | 33 ++-
.../config/MySqlIncrementalSourceOptions.java | 21 ++
.../cdc/mysql/source/MySqlIncrementalSource.java | 144 +++++++++
.../source/MySqlIncrementalSourceFactory.java | 12 +-
.../MySqlIncrementalSourceStartupConfigTest.java | 322 +++++++++++++++++++++
.../state/MySqlIncrementalSplitStateTest.java | 77 +++++
.../mysql/MysqlCDCSpecificStartingOffsetIT.java | 94 +++++-
.../resources/mysqlcdc_specific_gtid_offset.conf | 53 ++++
10 files changed, 752 insertions(+), 10 deletions(-)
diff --git a/docs/en/connectors/source/MySQL-CDC.md
b/docs/en/connectors/source/MySQL-CDC.md
index 87dc7d55c6..8ce4b5279a 100644
--- a/docs/en/connectors/source/MySQL-CDC.md
+++ b/docs/en/connectors/source/MySQL-CDC.md
@@ -196,6 +196,9 @@ When an initial consistent snapshot is made for large
databases, your establishe
| startup.mode | Enum | No | INITIAL |
Optional startup mode for MySQL CDC consumer, valid enumerations are `initial`,
`earliest`, `latest` , `specific` and `timestamp`. <br/> `initial`: Synchronize
historical data at startup, and then synchronize incremental data.<br/>
`earliest`: Startup from the earliest offset possible.<br/> `latest`: Startup
from the latest offset.<br/> `specific`: Startup from user-supplied specific
offsets.<br/> `timestamp`: [...]
| startup.specific-offset.file | String | No | - |
Start from the specified binlog file name. **Note, This option is required when
the `startup.mode` option used `specific`.**
[...]
| startup.specific-offset.pos | Long | No | - |
Start from the specified binlog file position. **Note, This option is required
when the `startup.mode` option used `specific`.**
[...]
+| startup.specific-offset.gtid-set | String | No | - |
Optional MySQL GTID set for `specific` startup mode. This option is used
together with `startup.specific-offset.file` and `startup.specific-offset.pos`.
[...]
+| startup.specific-offset.skip-events | Long | No | 0 |
Number of binlog events to skip after the configured specific startup offset.
This option can only be used when `startup.mode` is `specific`.
[...]
+| startup.specific-offset.skip-rows | Long | No | 0 |
Number of rows to skip after the configured specific startup offset. This
option can only be used when `startup.mode` is `specific`.
[...]
| startup.timestamp | Long | No | - |
Start from the specified timestamp, in milliseconds since Unix epoch. **Note,
This option is required when the `startup.mode` option uses `timestamp`.**
[...]
| stop.mode | Enum | No | NEVER |
Optional stop mode for MySQL CDC consumer, valid enumerations are `never`,
`latest` or `specific`. <br/> `never`: Real-time job don't stop the
source.<br/> `latest`: Stop from the latest offset.<br/> `specific`: Stop from
user-supplied specific offset.
[...]
| stop.specific-offset.file | String | No | - |
Stop from the specified binlog file name. **Note, This option is required when
the `stop.mode` option used `specific`.**
[...]
diff --git a/docs/zh/connectors/source/MySQL-CDC.md
b/docs/zh/connectors/source/MySQL-CDC.md
index 0f5ba675b3..21f71cf990 100644
--- a/docs/zh/connectors/source/MySQL-CDC.md
+++ b/docs/zh/connectors/source/MySQL-CDC.md
@@ -195,6 +195,9 @@ show variables where variable_name in ('log_bin',
'binlog_format', 'binlog_row_i
| startup.mode | Enum | 否 | INITIAL |
MySQL CDC 消费者的可选启动模式, 有效枚举值为 `initial`, `earliest`, `latest` , `specific` 和
`timestamp`. <br/> `initial`: 启动时同步历史数据, 然后同步增量数据.<br/> `earliest`:
从尽可能最早的偏移量开始启动.<br/> `latest`: 从最近的偏移量启动.<br/> `specific`:
从用户提供的特定偏移量开始启动.<br/> `timestamp`: 从用户提供的特定时间戳开始启动. |
| startup.specific-offset.file | String | 否 | - |
从指定的binlog日志文件名开始. **注意, 当使用 `startup.mode` 选项为 `specific` 时，此选项为必填项.**
|
| startup.specific-offset.pos | Long | 否 | - |
从指定的binlog日志文件位置开始. **注意, 当使用 `startup.mode` 选项为 `specific` 时，此选项为必填项.**
|
+| startup.specific-offset.gtid-set | String | 否 | - | 当
`startup.mode` 为 `specific` 时，可选配置 MySQL GTID 集合. 该选项需要和
`startup.specific-offset.file`、`startup.specific-offset.pos` 一起使用.
|
+| startup.specific-offset.skip-events | Long | 否 | 0 | 配置
specific 启动偏移量后需要跳过的 binlog event 数量. 该选项只能在 `startup.mode` 为 `specific` 时使用.
|
+| startup.specific-offset.skip-rows | Long | 否 | 0 | 配置
specific 启动偏移量后需要跳过的行数. 该选项只能在 `startup.mode` 为 `specific` 时使用.
|
| startup.timestamp | Long | 否 | - |
从指定时间戳启动，单位为 Unix 纪元以来的毫秒数。**注意，当 `startup.mode` 为 `timestamp` 时，此选项必填。**
|
| stop.mode | Enum | 否 | NEVER |
MySQL CDC 消费者的可选停止模式, 有效枚举值为 `never`, `latest` 和 `specific`. <br/> `never`:
实时任务一直运行不停止.<br/> `latest`: 从最新的偏移量处停止.<br/> `specific`: 从用户提供的特定偏移量处停止.
|
| stop.specific-offset.file | String | 否 | - |
从指定的binlog日志文件名停止. **注意, 当使用 `stop.mode` 选项为 `specific` 时，此选项为必填项.**
|
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/config/StartupConfig.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/config/StartupConfig.java
index 40dc755540..0f07bc4670 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/config/StartupConfig.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/config/StartupConfig.java
@@ -21,13 +21,13 @@ import
org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset;
import org.apache.seatunnel.connectors.cdc.base.source.offset.OffsetFactory;
-import lombok.AllArgsConstructor;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import java.io.Serializable;
+import java.util.LinkedHashMap;
+import java.util.Map;
-@AllArgsConstructor
@EqualsAndHashCode
public final class StartupConfig implements Serializable {
private static final long serialVersionUID = 1L;
@@ -35,6 +35,32 @@ public final class StartupConfig implements Serializable {
private final String specificOffsetFile;
private final Long specificOffsetPos;
@Getter private final Long timestamp;
+ private final Map<String, String> specificOffset;
+
+ public StartupConfig(
+ StartupMode startupMode,
+ String specificOffsetFile,
+ Long specificOffsetPos,
+ Long timestamp) {
+ this(startupMode, specificOffsetFile, specificOffsetPos, timestamp,
null);
+ }
+
+ public StartupConfig(StartupMode startupMode, Map<String, String>
specificOffset) {
+ this(startupMode, null, null, null, specificOffset);
+ }
+
+ public StartupConfig(
+ StartupMode startupMode,
+ String specificOffsetFile,
+ Long specificOffsetPos,
+ Long timestamp,
+ Map<String, String> specificOffset) {
+ this.startupMode = startupMode;
+ this.specificOffsetFile = specificOffsetFile;
+ this.specificOffsetPos = specificOffsetPos;
+ this.timestamp = timestamp;
+ this.specificOffset = specificOffset == null ? null : new
LinkedHashMap<>(specificOffset);
+ }
public Offset getStartupOffset(OffsetFactory offsetFactory) {
switch (startupMode) {
@@ -45,6 +71,9 @@ public final class StartupConfig implements Serializable {
case INITIAL:
return null;
case SPECIFIC:
+ if (specificOffset != null) {
+ return offsetFactory.specific(specificOffset);
+ }
return offsetFactory.specific(specificOffsetFile,
specificOffsetPos);
case TIMESTAMP:
return offsetFactory.timestamp(timestamp);
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/config/MySqlIncrementalSourceOptions.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/config/MySqlIncrementalSourceOptions.java
index 58d67fe6d0..03dd26a6e9 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/config/MySqlIncrementalSourceOptions.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/config/MySqlIncrementalSourceOptions.java
@@ -37,6 +37,27 @@ public class MySqlIncrementalSourceOptions extends
JdbcSourceOptions implements
.withDescription(
"int type narrowing, if true, the tinyint(1) type
will be narrowed to the boolean type if without loss of precision. Support for
MySQL at now.");
+ public static final Option<String> STARTUP_SPECIFIC_OFFSET_GTID_SET =
+ Options.key("startup.specific-offset.gtid-set")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional GTID set used with file and position in
case of \"specific\" startup mode.");
+
+ public static final Option<Long> STARTUP_SPECIFIC_OFFSET_SKIP_EVENTS =
+ Options.key("startup.specific-offset.skip-events")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional number of binlog events to skip after
the specific startup offset.");
+
+ public static final Option<Long> STARTUP_SPECIFIC_OFFSET_SKIP_ROWS =
+ Options.key("startup.specific-offset.skip-rows")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional number of rows to skip after the
specific startup offset.");
+
public static final SingleChoiceOption<StartupMode> STARTUP_MODE =
Options.key(SourceOptions.STARTUP_MODE_KEY)
.singleChoice(
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSource.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSource.java
index abb39f5897..7a12494447 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSource.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSource.java
@@ -17,6 +17,8 @@
package org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source;
+import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
+
import org.apache.seatunnel.api.configuration.Option;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.source.SupportParallelism;
@@ -27,8 +29,10 @@ import org.apache.seatunnel.common.utils.JdbcUrlUtil;
import org.apache.seatunnel.common.utils.SeaTunnelException;
import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
import org.apache.seatunnel.connectors.cdc.base.config.SourceConfig;
+import org.apache.seatunnel.connectors.cdc.base.config.StartupConfig;
import org.apache.seatunnel.connectors.cdc.base.dialect.DataSourceDialect;
import org.apache.seatunnel.connectors.cdc.base.option.JdbcSourceOptions;
+import org.apache.seatunnel.connectors.cdc.base.option.SourceOptions;
import org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
import org.apache.seatunnel.connectors.cdc.base.option.StopMode;
import org.apache.seatunnel.connectors.cdc.base.schema.SchemaChangeEventFilter;
@@ -41,17 +45,20 @@ import
org.apache.seatunnel.connectors.cdc.debezium.row.DebeziumJsonDeserializeS
import
org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializeSchema;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.config.MySqlIncrementalSourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.config.MySqlSourceConfigFactory;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.offset.BinlogOffset;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.offset.BinlogOffsetFactory;
import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions;
import org.apache.kafka.connect.data.Struct;
+import io.debezium.connector.mysql.GtidSet;
import io.debezium.jdbc.JdbcConnection;
import io.debezium.relational.TableId;
import io.debezium.relational.history.TableChanges;
import java.time.ZoneId;
import java.util.Arrays;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -67,6 +74,143 @@ public class MySqlIncrementalSource<T> extends
IncrementalSource<T, JdbcSourceCo
super(options, catalogTables);
}
+ @Override
+ protected StartupConfig getStartupConfig(ReadonlyConfig config) {
+ return createStartupConfig(config);
+ }
+
+ /* Route MySQL specific startup through the map-based offset path for GTID
and skip metadata. */
+ static StartupConfig createStartupConfig(ReadonlyConfig config) {
+ StartupMode startupMode =
config.get(MySqlIncrementalSourceOptions.STARTUP_MODE);
+ if (StartupMode.SPECIFIC.equals(startupMode)) {
+ return new StartupConfig(startupMode,
createSpecificStartupOffset(config));
+ }
+
+ validateNoSpecificStartupOffset(config, startupMode);
+ return new StartupConfig(
+ startupMode,
+ config.get(SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE),
+ config.get(SourceOptions.STARTUP_SPECIFIC_OFFSET_POS),
+ config.get(SourceOptions.STARTUP_TIMESTAMP));
+ }
+
+ /*
+ * Debezium's MySqlOffsetContext.Loader requires file and pos for a
specific startup offset.
+ * GTID and skip fields are optional metadata on that same offset, not an
alternative anchor.
+ */
+ private static Map<String, String>
createSpecificStartupOffset(ReadonlyConfig config) {
+ Optional<String> gtidSet = getValidatedGtidSet(config);
+ boolean hasFile =
+
config.getOptional(SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE).isPresent();
+ boolean hasPos =
config.getOptional(SourceOptions.STARTUP_SPECIFIC_OFFSET_POS).isPresent();
+
+ if (hasFile != hasPos) {
+ throw new IllegalArgumentException(
+ String.format(
+ "'%s' and '%s' must be configured together when
'%s' is 'specific'.",
+ SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE.key(),
+ SourceOptions.STARTUP_SPECIFIC_OFFSET_POS.key(),
+ SourceOptions.STARTUP_MODE_KEY));
+ }
+
+ if (!hasFile) {
+ throw new IllegalArgumentException(
+ String.format(
+ "'%s' requires '%s' with '%s' when the mode is
'specific'.",
+ SourceOptions.STARTUP_MODE_KEY,
+ SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE.key(),
+ SourceOptions.STARTUP_SPECIFIC_OFFSET_POS.key()));
+ }
+
+ long skipEvents =
+ getNonNegativeSkipValue(
+ config,
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_EVENTS);
+ long skipRows =
+ getNonNegativeSkipValue(
+ config,
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_ROWS);
+
+ Map<String, String> offset = new LinkedHashMap<>();
+ String file = config.get(SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE);
+ if (StringUtils.isBlank(file)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "'%s' must not be blank.",
+ SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE.key()));
+ }
+ offset.put(BinlogOffset.BINLOG_FILENAME_OFFSET_KEY, file);
+ offset.put(
+ BinlogOffset.BINLOG_POSITION_OFFSET_KEY,
+
String.valueOf(config.get(SourceOptions.STARTUP_SPECIFIC_OFFSET_POS)));
+ if (gtidSet.isPresent()) {
+ offset.put(BinlogOffset.GTID_SET_KEY, gtidSet.get());
+ }
+ offset.put(BinlogOffset.EVENTS_TO_SKIP_OFFSET_KEY,
String.valueOf(skipEvents));
+ offset.put(BinlogOffset.ROWS_TO_SKIP_OFFSET_KEY,
String.valueOf(skipRows));
+ return offset;
+ }
+
+ /* Validate the configured GTID set before adding it to Debezium's offset
map. */
+ private static Optional<String> getValidatedGtidSet(ReadonlyConfig config)
{
+ Optional<String> configuredGtidSet =
+
config.getOptional(MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_GTID_SET);
+ if (!configuredGtidSet.isPresent()) {
+ return Optional.empty();
+ }
+
+ String gtidSet = configuredGtidSet.get().trim();
+ if (StringUtils.isBlank(gtidSet)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "'%s' must not be blank.",
+
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_GTID_SET.key()));
+ }
+
+ try {
+ new GtidSet(gtidSet);
+ } catch (Exception e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Invalid '%s' value '%s'.",
+
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_GTID_SET.key(),
+ gtidSet),
+ e);
+ }
+ return Optional.of(gtidSet);
+ }
+
+ private static long getNonNegativeSkipValue(ReadonlyConfig config,
Option<Long> option) {
+ long value = config.getOptional(option).orElse(0L);
+ if (value < 0) {
+ throw new IllegalArgumentException(
+ String.format("'%s' must be greater than or equal to 0.",
option.key()));
+ }
+ return value;
+ }
+
+ private static void validateNoSpecificStartupOffset(
+ ReadonlyConfig config, StartupMode startupMode) {
+ if (hasSpecificStartupOffset(config)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "'startup.specific-offset.*' options can only be
used when '%s' is 'specific', but current mode is '%s'.",
+ SourceOptions.STARTUP_MODE_KEY, startupMode));
+ }
+ }
+
+ private static boolean hasSpecificStartupOffset(ReadonlyConfig config) {
+ return
config.getOptional(SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE).isPresent()
+ ||
config.getOptional(SourceOptions.STARTUP_SPECIFIC_OFFSET_POS).isPresent()
+ || config.getOptional(
+
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_GTID_SET)
+ .isPresent()
+ || config.getOptional(
+
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_EVENTS)
+ .isPresent()
+ || config.getOptional(
+
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_ROWS)
+ .isPresent();
+ }
+
@Override
public Option<StartupMode> getStartupModeOption() {
return MySqlIncrementalSourceOptions.STARTUP_MODE;
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactory.java
index 04858dcb29..9f4a45fc04 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactory.java
@@ -80,7 +80,12 @@ public class MySqlIncrementalSourceFactory extends
BaseChangeStreamTableSourceFa
MySqlIncrementalSourceOptions.SCHEMA_CHANGES_ENABLED,
MySqlIncrementalSourceOptions.SCHEMA_CHANGES_INCLUDE,
MySqlIncrementalSourceOptions.SCHEMA_CHANGES_EXCLUDE,
- MySqlIncrementalSourceOptions.INT_TYPE_NARROWING)
+ MySqlIncrementalSourceOptions.INT_TYPE_NARROWING,
+ SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE,
+ SourceOptions.STARTUP_SPECIFIC_OFFSET_POS,
+
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_GTID_SET,
+
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_EVENTS,
+
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_ROWS)
.optional(
MySqlIncrementalSourceOptions.STARTUP_MODE,
MySqlIncrementalSourceOptions.STOP_MODE)
@@ -88,11 +93,6 @@ public class MySqlIncrementalSourceFactory extends
BaseChangeStreamTableSourceFa
MySqlIncrementalSourceOptions.STARTUP_MODE,
StartupMode.INITIAL,
SourceOptions.EXACTLY_ONCE)
- .conditional(
- MySqlIncrementalSourceOptions.STARTUP_MODE,
- StartupMode.SPECIFIC,
- SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE,
- SourceOptions.STARTUP_SPECIFIC_OFFSET_POS)
.conditional(
MySqlIncrementalSourceOptions.STOP_MODE,
StopMode.SPECIFIC,
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceStartupConfigTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceStartupConfigTest.java
new file mode 100644
index 0000000000..b546e64e76
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceStartupConfigTest.java
@@ -0,0 +1,322 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConfigValidator;
+import org.apache.seatunnel.connectors.cdc.base.config.StartupConfig;
+import org.apache.seatunnel.connectors.cdc.base.option.SourceOptions;
+import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset;
+import org.apache.seatunnel.connectors.cdc.base.source.offset.OffsetFactory;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.config.MySqlIncrementalSourceOptions;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.offset.BinlogOffset;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import io.debezium.config.Configuration;
+import io.debezium.connector.mysql.MySqlConnectorConfig;
+import io.debezium.connector.mysql.MySqlOffsetContext;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class MySqlIncrementalSourceStartupConfigTest {
+
+ private static final String GTID_SET =
"3E11FA47-71CA-11E1-9E33-C80AA9429562:1-10";
+
+ @Test
+ public void testOptionRuleAcceptsGtidSpecificStartup() {
+ Map<String, Object> options = requiredOptions();
+ options.put(SourceOptions.STARTUP_MODE_KEY, "specific");
+ options.put(SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE.key(),
"mysql-bin.000123");
+ options.put(SourceOptions.STARTUP_SPECIFIC_OFFSET_POS.key(), 456789L);
+
options.put(MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_GTID_SET.key(),
GTID_SET);
+
+ ConfigValidator.of(ReadonlyConfig.fromMap(options))
+ .validate(new MySqlIncrementalSourceFactory().optionRule());
+ }
+
+ @Test
+ public void testCreateStartupConfigWithGtidSetAndSkipFields() {
+ StartupConfig startupConfig =
+ MySqlIncrementalSource.createStartupConfig(
+ config(
+ SourceOptions.STARTUP_MODE_KEY,
+ "specific",
+
SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE.key(),
+ "mysql-bin.000123",
+
SourceOptions.STARTUP_SPECIFIC_OFFSET_POS.key(),
+ 456789L,
+
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_GTID_SET
+ .key(),
+ GTID_SET,
+
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_EVENTS
+ .key(),
+ 3L,
+
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_ROWS
+ .key(),
+ 10L));
+
+ Offset startupOffset = startupConfig.getStartupOffset(new
TestOffsetFactory());
+
+ Assertions.assertEquals(
+ "mysql-bin.000123",
+
startupOffset.getOffset().get(BinlogOffset.BINLOG_FILENAME_OFFSET_KEY));
+ Assertions.assertEquals(
+ "456789",
startupOffset.getOffset().get(BinlogOffset.BINLOG_POSITION_OFFSET_KEY));
+ Assertions.assertEquals(GTID_SET,
startupOffset.getOffset().get(BinlogOffset.GTID_SET_KEY));
+ Assertions.assertEquals(
+ "3",
startupOffset.getOffset().get(BinlogOffset.EVENTS_TO_SKIP_OFFSET_KEY));
+ Assertions.assertEquals(
+ "10",
startupOffset.getOffset().get(BinlogOffset.ROWS_TO_SKIP_OFFSET_KEY));
+ }
+
+ @Test
+ public void testCreateStartupConfigWithFilePositionAndSkipFields() {
+ StartupConfig startupConfig =
+ MySqlIncrementalSource.createStartupConfig(
+ config(
+ SourceOptions.STARTUP_MODE_KEY,
+ "specific",
+
SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE.key(),
+ "mysql-bin.000123",
+
SourceOptions.STARTUP_SPECIFIC_OFFSET_POS.key(),
+ 456789L,
+
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_ROWS
+ .key(),
+ 2L));
+
+ Offset startupOffset = startupConfig.getStartupOffset(new
TestOffsetFactory());
+
+ Assertions.assertEquals(
+ "mysql-bin.000123",
+
startupOffset.getOffset().get(BinlogOffset.BINLOG_FILENAME_OFFSET_KEY));
+ Assertions.assertEquals(
+ "456789",
startupOffset.getOffset().get(BinlogOffset.BINLOG_POSITION_OFFSET_KEY));
+ Assertions.assertEquals(
+ "0",
startupOffset.getOffset().get(BinlogOffset.EVENTS_TO_SKIP_OFFSET_KEY));
+ Assertions.assertEquals(
+ "2",
startupOffset.getOffset().get(BinlogOffset.ROWS_TO_SKIP_OFFSET_KEY));
+ }
+
+ @Test
+ public void testSpecificStartupOffsetLoadsThroughDebeziumOffsetLoader() {
+ StartupConfig startupConfig =
+ MySqlIncrementalSource.createStartupConfig(
+ config(
+ SourceOptions.STARTUP_MODE_KEY,
+ "specific",
+
SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE.key(),
+ "mysql-bin.000123",
+
SourceOptions.STARTUP_SPECIFIC_OFFSET_POS.key(),
+ 456789L,
+
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_GTID_SET
+ .key(),
+ GTID_SET,
+
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_EVENTS
+ .key(),
+ 3L,
+
MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_ROWS
+ .key(),
+ 10L));
+
+ Offset startupOffset = startupConfig.getStartupOffset(new
TestOffsetFactory());
+ MySqlOffsetContext offsetContext =
+ new
MySqlOffsetContext.Loader(debeziumConfig()).load(startupOffset.getOffset());
+
+ Assertions.assertEquals("mysql-bin.000123",
offsetContext.getSource().binlogFilename());
+ Assertions.assertEquals(456789L,
offsetContext.getSource().binlogPosition());
+ Assertions.assertEquals(GTID_SET, offsetContext.gtidSet());
+ Assertions.assertEquals(3L, offsetContext.eventsToSkipUponRestart());
+ Assertions.assertEquals(10, offsetContext.rowsToSkipUponRestart());
+ }
+
+ @Test
+ public void testCreateStartupConfigRejectsGtidSetWithoutFilePosition() {
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ MySqlIncrementalSource.createStartupConfig(
+ config(
+ SourceOptions.STARTUP_MODE_KEY,
+ "specific",
+ MySqlIncrementalSourceOptions
+
.STARTUP_SPECIFIC_OFFSET_GTID_SET
+ .key(),
+ GTID_SET)));
+
+ Assertions.assertTrue(
+
exception.getMessage().contains(SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE.key()));
+ }
+
+ @Test
+ public void testCreateStartupConfigRejectsMissingSpecificOffset() {
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ MySqlIncrementalSource.createStartupConfig(
+ config(SourceOptions.STARTUP_MODE_KEY,
"specific")));
+
+ Assertions.assertTrue(exception.getMessage().contains("requires"));
+ }
+
+ @Test
+ public void testCreateStartupConfigRejectsSkipOutsideSpecificMode() {
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ MySqlIncrementalSource.createStartupConfig(
+ config(
+ SourceOptions.STARTUP_MODE_KEY,
+ "latest",
+ MySqlIncrementalSourceOptions
+
.STARTUP_SPECIFIC_OFFSET_SKIP_EVENTS
+ .key(),
+ 1L)));
+
+
Assertions.assertTrue(exception.getMessage().contains("startup.specific-offset.*"));
+ }
+
+ @Test
+ public void testCreateStartupConfigRejectsMalformedGtidSet() {
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ MySqlIncrementalSource.createStartupConfig(
+ config(
+ SourceOptions.STARTUP_MODE_KEY,
+ "specific",
+
SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE.key(),
+ "mysql-bin.000123",
+
SourceOptions.STARTUP_SPECIFIC_OFFSET_POS.key(),
+ 456789L,
+ MySqlIncrementalSourceOptions
+
.STARTUP_SPECIFIC_OFFSET_GTID_SET
+ .key(),
+ "bad-gtid")));
+
+ Assertions.assertTrue(exception.getMessage().contains("Invalid"));
+ }
+
+ @Test
+ public void testCreateStartupConfigRejectsNegativeSkipRows() {
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ MySqlIncrementalSource.createStartupConfig(
+ config(
+ SourceOptions.STARTUP_MODE_KEY,
+ "specific",
+
SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE.key(),
+ "mysql-bin.000123",
+
SourceOptions.STARTUP_SPECIFIC_OFFSET_POS.key(),
+ 456789L,
+ MySqlIncrementalSourceOptions
+
.STARTUP_SPECIFIC_OFFSET_GTID_SET
+ .key(),
+ GTID_SET,
+ MySqlIncrementalSourceOptions
+
.STARTUP_SPECIFIC_OFFSET_SKIP_ROWS
+ .key(),
+ -1L)));
+
+ Assertions.assertTrue(exception.getMessage().contains("greater than or
equal to 0"));
+ }
+
+ private static ReadonlyConfig config(Object... keysAndValues) {
+ Map<String, Object> options = new HashMap<>();
+ for (int i = 0; i < keysAndValues.length; i += 2) {
+ options.put((String) keysAndValues[i], keysAndValues[i + 1]);
+ }
+ return ReadonlyConfig.fromMap(options);
+ }
+
+ private static MySqlConnectorConfig debeziumConfig() {
+ return new MySqlConnectorConfig(
+ Configuration.create()
+ .with(MySqlConnectorConfig.SERVER_NAME, "test_server")
+ .with(MySqlConnectorConfig.HOSTNAME, "localhost")
+ .with(MySqlConnectorConfig.USER, "test")
+ .with(MySqlConnectorConfig.PASSWORD, "test")
+ .build());
+ }
+
+ private static Map<String, Object> requiredOptions() {
+ Map<String, Object> options = new HashMap<>();
+ options.put(MySqlIncrementalSourceOptions.USERNAME.key(), "mysqluser");
+ options.put(MySqlIncrementalSourceOptions.PASSWORD.key(), "mysqlpw");
+ options.put(MySqlIncrementalSourceOptions.URL.key(),
"jdbc:mysql://localhost:3306/test");
+ options.put(
+ MySqlIncrementalSourceOptions.TABLE_NAMES.key(),
+ Collections.singletonList("test.table1"));
+ return options;
+ }
+
+ private static class TestOffsetFactory extends OffsetFactory {
+ @Override
+ public Offset earliest() {
+ return new TestOffset(Collections.emptyMap());
+ }
+
+ @Override
+ public Offset neverStop() {
+ return new TestOffset(Collections.emptyMap());
+ }
+
+ @Override
+ public Offset latest() {
+ return new TestOffset(Collections.emptyMap());
+ }
+
+ @Override
+ public Offset specific(Map<String, String> offset) {
+ return new TestOffset(offset);
+ }
+
+ @Override
+ public Offset specific(String filename, Long position) {
+ Map<String, String> offset = new HashMap<>();
+ offset.put(BinlogOffset.BINLOG_FILENAME_OFFSET_KEY, filename);
+ offset.put(BinlogOffset.BINLOG_POSITION_OFFSET_KEY,
String.valueOf(position));
+ return new TestOffset(offset);
+ }
+
+ @Override
+ public Offset timestamp(long timestamp) {
+ return new TestOffset(Collections.emptyMap());
+ }
+ }
+
+ private static class TestOffset extends Offset {
+ TestOffset(Map<String, String> offset) {
+ this.offset = offset;
+ }
+
+ @Override
+ public int compareTo(Offset offset) {
+ return 0;
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/split/state/MySqlIncrementalSplitStateTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/split/state/MySqlIncrementalSplitStateTest.java
new file mode 100644
index 0000000000..2b33984e27
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/split/state/MySqlIncrementalSplitStateTest.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.split.state;
+
+import org.apache.seatunnel.connectors.cdc.base.source.split.IncrementalSplit;
+import
org.apache.seatunnel.connectors.cdc.base.source.split.state.IncrementalSplitState;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.offset.BinlogOffset;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import io.debezium.relational.TableId;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class MySqlIncrementalSplitStateTest {
+
+ private static final String GTID_SET =
"3E11FA47-71CA-11E1-9E33-C80AA9429562:1-10";
+
+ @Test
+ public void testToSourceSplitPreservesGtidAndSkipStartupOffset() {
+ BinlogOffset startupOffset = gtidOffset(3L, 10L);
+ IncrementalSplit split =
+ new IncrementalSplit(
+ "incremental-split",
+ Collections.singletonList(new TableId("test_db", null,
"test_table")),
+ startupOffset,
+ null,
+ Collections.emptyList(),
+ null,
+ Collections.emptyMap());
+
+ IncrementalSplit restoredSplit = new
IncrementalSplitState(split).toSourceSplit();
+
+ Assertions.assertEquals(startupOffset,
restoredSplit.getStartupOffset());
+ Assertions.assertEquals(
+ GTID_SET,
+
restoredSplit.getStartupOffset().getOffset().get(BinlogOffset.GTID_SET_KEY));
+ Assertions.assertEquals(
+ "3",
+ restoredSplit
+ .getStartupOffset()
+ .getOffset()
+ .get(BinlogOffset.EVENTS_TO_SKIP_OFFSET_KEY));
+ Assertions.assertEquals(
+ "10",
+ restoredSplit
+ .getStartupOffset()
+ .getOffset()
+ .get(BinlogOffset.ROWS_TO_SKIP_OFFSET_KEY));
+ }
+
+ private static BinlogOffset gtidOffset(long skipEvents, long skipRows) {
+ Map<String, String> offset = new HashMap<>();
+ offset.put(BinlogOffset.GTID_SET_KEY, GTID_SET);
+ offset.put(BinlogOffset.EVENTS_TO_SKIP_OFFSET_KEY,
String.valueOf(skipEvents));
+ offset.put(BinlogOffset.ROWS_TO_SKIP_OFFSET_KEY,
String.valueOf(skipRows));
+ return new BinlogOffset(offset);
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCSpecificStartingOffsetIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCSpecificStartingOffsetIT.java
index 66762b64d1..5598879257 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCSpecificStartingOffsetIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCSpecificStartingOffsetIT.java
@@ -62,6 +62,7 @@ import java.util.Collections;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -177,7 +178,9 @@ public class MysqlCDCSpecificStartingOffsetIT extends
TestSuiteBase implements T
throws IOException, InterruptedException {
String jobId = String.valueOf(JobIdGenerator.newJobId());
String jobConfigFile = "/mysqlcdc_earliest_offset.conf";
- purgeBinaryLogs();
+ clearTable(MYSQL_DATABASE, SOURCE_TABLE_1);
+ clearTable(MYSQL_DATABASE, SINK_TABLE);
+ rotateAndPurgeBinaryLogs();
// Insert data
executeSql(
String.format(
@@ -274,7 +277,7 @@ public class MysqlCDCSpecificStartingOffsetIT extends
TestSuiteBase implements T
public void testMysqlCdcSpecificOffset(TestContainer container) throws
Exception {
String jobId = String.valueOf(JobIdGenerator.newJobId());
String jobConfigFile = "/mysqlcdc_specific_offset.conf";
- purgeBinaryLogs();
+ rotateAndPurgeBinaryLogs();
String source_sql_where_id_template =
"select id, cast(f_binary as char) as f_binary, cast(f_blob as
char) as f_blob, cast(f_long_varbinary as char) as f_long_varbinary,"
+ " cast(f_longblob as char) as f_longblob,
cast(f_tinyblob as char) as f_tinyblob, cast(f_varbinary as char) as
f_varbinary,"
@@ -396,6 +399,88 @@ public class MysqlCDCSpecificStartingOffsetIT extends
TestSuiteBase implements T
});
}
+ @TestTemplate
+ public void testMysqlCdcSpecificGtidOffsetWithSkipRows(TestContainer
container)
+ throws Exception {
+ String jobId = String.valueOf(JobIdGenerator.newJobId());
+ String jobConfigFile = "/mysqlcdc_specific_gtid_offset.conf";
+ String sourceSqlWhereIdTemplate =
+ "select id, cast(f_binary as char) as f_binary, cast(f_blob as
char) as f_blob, cast(f_long_varbinary as char) as f_long_varbinary,"
+ + " cast(f_longblob as char) as f_longblob,
cast(f_tinyblob as char) as f_tinyblob, cast(f_varbinary as char) as
f_varbinary,"
+ + " f_smallint, f_smallint_unsigned, f_mediumint,
f_mediumint_unsigned, f_int, f_int_unsigned, f_integer, f_integer_unsigned,"
+ + " f_bigint, f_bigint_unsigned, f_numeric, f_decimal,
f_float, f_double, f_double_precision, f_longtext, f_mediumtext,"
+ + " f_text, f_tinytext, f_varchar, f_date, f_datetime,
f_timestamp, f_bit1, cast(f_bit64 as char) as f_bit64, f_char,"
+ + " f_enum, cast(f_mediumblob as char) as
f_mediumblob, f_long_varchar, f_real, f_time, f_tinyint, f_tinyint_unsigned,"
+ + " f_json, f_year from %s.%s where id in (%s)";
+
+ clearTable(MYSQL_DATABASE, SOURCE_TABLE_1);
+ clearTable(MYSQL_DATABASE, SINK_TABLE);
+ rotateAndPurgeBinaryLogs();
+
+ BinlogOffset binlogOffset = getCurrentBinlogOffset();
+ String gtidSet = binlogOffset.getGtidSet();
+ Assertions.assertNotNull(gtidSet);
+ Assertions.assertFalse(gtidSet.trim().isEmpty());
+ String[] variables = {
+ "specific_offset_file=" + binlogOffset.getFilename(),
+ "specific_offset_pos=" + binlogOffset.getPosition(),
+ "specific_offset_gtid_set=" + gtidSet,
+ "specific_offset_skip_rows=1"
+ };
+
+ executeSql(
+ String.format(
+ "INSERT INTO %s.%s (id) VALUES (16), (17)",
+ MYSQL_DATABASE, SOURCE_TABLE_1));
+
+ CompletableFuture<Container.ExecResult> jobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.executeJob(jobConfigFile,
jobId, variables);
+ } catch (Exception e) {
+ log.error("Commit task exception :" +
e.getMessage());
+ throw new RuntimeException(e);
+ }
+ });
+
+ await().atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () -> {
+ assertJobHasNotFinished(jobFuture);
+ Assertions.assertIterableEquals(
+ query(
+ String.format(
+ sourceSqlWhereIdTemplate,
+ MYSQL_DATABASE,
+ SOURCE_TABLE_1,
+ "17")),
+ query(getSinkQuerySQL(MYSQL_DATABASE,
SINK_TABLE)));
+ });
+
+ Container.ExecResult cancelResult = container.cancelJob(jobId);
+ Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
+
+ Container.ExecResult result = jobFuture.get(60, TimeUnit.SECONDS);
+ Assertions.assertEquals(0, result.getExitCode(), result.getStderr());
+ }
+
+ private static void
assertJobHasNotFinished(CompletableFuture<Container.ExecResult> jobFuture) {
+ if (!jobFuture.isDone()) {
+ return;
+ }
+ try {
+ Container.ExecResult result = jobFuture.getNow(null);
+ Assertions.fail(
+ "Streaming job finished before the sink assertion.
exitCode="
+ + result.getExitCode()
+ + ", stderr="
+ + result.getStderr());
+ } catch (CompletionException e) {
+ Assertions.fail("Streaming job failed before the sink assertion.",
e);
+ }
+ }
+
@TestTemplate
public void testMysqlCdcTimestampOffset(TestContainer container) throws
Exception {
log.info("begin testMysqlCdcTimestampOffset");
@@ -575,6 +660,11 @@ public class MysqlCDCSpecificStartingOffsetIT extends
TestSuiteBase implements T
executeSql("FLUSH LOGS;");
}
+ private void rotateAndPurgeBinaryLogs() {
+ flushLogs();
+ purgeBinaryLogs();
+ }
+
private String getSourceQuerySQL(String database, String tableName) {
return String.format(SOURCE_SQL_TEMPLATE, database, tableName);
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_specific_gtid_offset.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_specific_gtid_offset.conf
new file mode 100644
index 0000000000..d45a905c3e
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_specific_gtid_offset.conf
@@ -0,0 +1,53 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ MySQL-CDC {
+ plugin_output = "customers_mysql_cdc"
+ server-id = 5654
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["mysql_cdc.mysql_cdc_e2e_source_table"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ startup.mode = "specific"
+ startup.specific-offset.file = ${specific_offset_file}
+ startup.specific-offset.pos = ${specific_offset_pos}
+ startup.specific-offset.gtid-set = ${specific_offset_gtid_set}
+ startup.specific-offset.skip-rows = ${specific_offset_skip_rows}
+ }
+}
+
+sink {
+ jdbc {
+ plugin_input = "customers_mysql_cdc"
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "st_user_sink"
+ password = "mysqlpw"
+
+ generate_sink_sql = true
+ database = mysql_cdc
+ table = mysql_cdc_e2e_sink_table
+ primary_keys = ["id"]
+ }
+}
