This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12708-83603e19b7b09d6114c6f6c9e89bfdafe2796183 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 262d623d9eddc49be3dc0184b39e540736bcce19 Author: Ryan Yang <[email protected]> AuthorDate: Mon Oct 12 05:22:29 2026 +0000 [Fix][Connector-V2][CDC] Validate startup.specific-offset.* options are only bound under STARTUP_MODE = SPECIFIC (#12708) --- .../mysql/config/MySqlSpecificOffsetGuards.java | 258 +++++++++++++++++++++ .../source/MySqlIncrementalSourceFactory.java | 59 ++++- .../source/MySqlIncrementalSourceFactoryTest.java | 203 ++++++++++++++++ .../OceanBaseIncrementalSourceFactoryTest.java | 45 ++++ 4 files changed, 557 insertions(+), 8 deletions(-) diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/config/MySqlSpecificOffsetGuards.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/config/MySqlSpecificOffsetGuards.java new file mode 100644 index 0000000000..b8f8f29654 --- /dev/null +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/config/MySqlSpecificOffsetGuards.java @@ -0,0 +1,258 @@ +/* + * 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.config; + +import org.apache.seatunnel.api.configuration.Option; +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.configuration.util.ConditionExtension; +import org.apache.seatunnel.api.configuration.util.OptionValidationException; +import org.apache.seatunnel.connectors.cdc.base.option.StartupMode; + +/** + * Reusable {@link ConditionExtension} guards for the MySQL CDC specific startup options. + * + * <p>Each guard is attached to one option of the source option rule, so a configuration that uses + * specific startup offsets is validated with the same rules the runtime source applies: the offsets + * are only accepted when the {@code startup.mode} is (or defaults to) {@code specific}, the binlog + * file and position must be configured together, and the value domains (non-blank / non-negative) + * match the runtime checks. The {@code startup.mode} option itself carries a guard as well, so + * {@code specific} without any anchor is rejected before the job starts. A valid anchor with + * optional GTID / skip metadata, and a configuration without any specific offset, remain accepted. + * + * <p>Runtime behavior is unchanged; the guards only surface the same failures before the job is + * started. The errors reuse the wording of the runtime error messages so that operators see the + * same message either way. + */ +public final class MySqlSpecificOffsetGuards { + + private MySqlSpecificOffsetGuards() {} + + public static ConditionExtension<String> specificOffsetFileGuard( + Option<StartupMode> startupModeOption, + Option<String> fileOption, + Option<Long> posOption) { + return new SpecificOffsetFileGuard(startupModeOption, fileOption.key(), posOption); + } + + public static ConditionExtension<Long> specificOffsetPosGuard( + Option<StartupMode> startupModeOption, + Option<Long> posOption, + Option<String> fileOption) { + return new SpecificOffsetPosGuard(startupModeOption, posOption.key(), fileOption); + } + + public static ConditionExtension<String> specificOffsetGtidSetGuard( + Option<StartupMode> startupModeOption, Option<String> gtidSetOption) { + return new SpecificOffsetGtidSetGuard(startupModeOption, gtidSetOption.key()); + } + + public static ConditionExtension<Long> specificOffsetSkipGuard( + Option<StartupMode> startupModeOption, Option<Long> skipOption) { + return new SpecificOffsetSkipGuard(startupModeOption, skipOption.key()); + } + + public static ConditionExtension<StartupMode> specificStartupModeGuard( + Option<StartupMode> startupModeOption, + Option<String> fileOption, + Option<Long> posOption) { + return new SpecificStartupModeGuard(startupModeOption, fileOption, posOption); + } + + protected abstract static class ModeSpecificGuard<T> implements ConditionExtension<T> { + protected final Option<StartupMode> startupModeOption; + protected final String guardedKey; + + protected ModeSpecificGuard(Option<StartupMode> startupModeOption, String guardedKey) { + this.startupModeOption = startupModeOption; + this.guardedKey = guardedKey; + } + + protected void checkStartupMode(ReadonlyConfig config) { + StartupMode mode = + config.getOptional(startupModeOption) + .orElseGet(startupModeOption::defaultValue); + if (mode != StartupMode.SPECIFIC) { + throw new OptionValidationException( + String.format( + "'startup.specific-offset.*' options can only be used when '%s' is" + + " 'specific', but current mode is '%s'.", + startupModeOption.key(), mode.name())); + } + } + + protected void checkPairedOption( + ReadonlyConfig config, Option<?> otherOption, String otherKey) { + if (!config.getOptional(otherOption).isPresent()) { + throw new OptionValidationException( + String.format( + "'%s' and '%s' must be configured together when '%s' is" + + " 'specific'.", + guardedKey, otherKey, startupModeOption.key())); + } + } + } + + /** + * Rejects {@code startup.mode = specific} when neither anchor of the specific offsets is + * configured. A partially configured pair is reported by the guard of the option that was + * configured, so this guard only covers the case where both anchors are absent. + */ + static final class SpecificStartupModeGuard implements ConditionExtension<StartupMode> { + private final Option<StartupMode> startupModeOption; + private final Option<String> fileOption; + private final Option<Long> posOption; + + SpecificStartupModeGuard( + Option<StartupMode> startupModeOption, + Option<String> fileOption, + Option<Long> posOption) { + this.startupModeOption = startupModeOption; + this.fileOption = fileOption; + this.posOption = posOption; + } + + @Override + public String description() { + return "requires '" + + fileOption.key() + + "' with '" + + posOption.key() + + "' when the mode is 'specific'"; + } + + @Override + public boolean evaluate(ReadonlyConfig config, StartupMode value) { + StartupMode mode = value != null ? value : startupModeOption.defaultValue(); + if (mode != StartupMode.SPECIFIC) { + return true; + } + if (!config.getOptional(fileOption).isPresent() + && !config.getOptional(posOption).isPresent()) { + throw new OptionValidationException( + String.format( + "'%s' requires '%s' with '%s' when the mode is 'specific'.", + startupModeOption.key(), fileOption.key(), posOption.key())); + } + return true; + } + } + + static final class SpecificOffsetFileGuard extends ModeSpecificGuard<String> { + private final Option<Long> posOption; + + SpecificOffsetFileGuard( + Option<StartupMode> startupModeOption, String guardedKey, Option<Long> posOption) { + super(startupModeOption, guardedKey); + this.posOption = posOption; + } + + @Override + public String description() { + return "can only be used when '" + + startupModeOption.key() + + "' is 'specific', must not be blank and must be paired with '" + + posOption.key() + + "'"; + } + + @Override + public boolean evaluate(ReadonlyConfig config, String value) { + checkStartupMode(config); + if (value.trim().isEmpty()) { + throw new OptionValidationException( + String.format("'%s' must not be blank.", guardedKey)); + } + checkPairedOption(config, posOption, posOption.key()); + return true; + } + } + + static final class SpecificOffsetPosGuard extends ModeSpecificGuard<Long> { + private final Option<String> fileOption; + + SpecificOffsetPosGuard( + Option<StartupMode> startupModeOption, + String guardedKey, + Option<String> fileOption) { + super(startupModeOption, guardedKey); + this.fileOption = fileOption; + } + + @Override + public String description() { + return "can only be used when '" + + startupModeOption.key() + + "' is 'specific' and must be paired with '" + + fileOption.key() + + "'"; + } + + @Override + public boolean evaluate(ReadonlyConfig config, Long value) { + checkStartupMode(config); + checkPairedOption(config, fileOption, fileOption.key()); + return true; + } + } + + static final class SpecificOffsetGtidSetGuard extends ModeSpecificGuard<String> { + SpecificOffsetGtidSetGuard(Option<StartupMode> startupModeOption, String guardedKey) { + super(startupModeOption, guardedKey); + } + + @Override + public String description() { + return "can only be used when '" + + startupModeOption.key() + + "' is 'specific' and must not be blank"; + } + + @Override + public boolean evaluate(ReadonlyConfig config, String value) { + checkStartupMode(config); + if (value.trim().isEmpty()) { + throw new OptionValidationException( + String.format("'%s' must not be blank.", guardedKey)); + } + return true; + } + } + + static final class SpecificOffsetSkipGuard extends ModeSpecificGuard<Long> { + SpecificOffsetSkipGuard(Option<StartupMode> startupModeOption, String guardedKey) { + super(startupModeOption, guardedKey); + } + + @Override + public String description() { + return "can only be used when '" + + startupModeOption.key() + + "' is 'specific' and must be greater than or equal to 0"; + } + + @Override + public boolean evaluate(ReadonlyConfig config, Long value) { + checkStartupMode(config); + if (value < 0L) { + throw new OptionValidationException( + String.format("'%s' must be greater than or equal to 0.", guardedKey)); + } + return true; + } + } +} 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 b16b0ddfc1..3d4d244e72 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 @@ -18,6 +18,7 @@ package org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source; import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.configuration.util.Conditions; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.source.SeaTunnelSource; import org.apache.seatunnel.api.source.SourceSplit; @@ -36,6 +37,7 @@ import org.apache.seatunnel.connectors.cdc.base.source.BaseChangeStreamTableSour import org.apache.seatunnel.connectors.cdc.base.utils.CatalogTableUtils; 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.config.MySqlSpecificOffsetGuards; import com.google.auto.service.AutoService; import lombok.extern.slf4j.Slf4j; @@ -90,15 +92,16 @@ public class MySqlIncrementalSourceFactory extends BaseChangeStreamTableSourceFa MySqlIncrementalSourceOptions.SCHEMA_CHANGES_ENABLED, MySqlIncrementalSourceOptions.SCHEMA_CHANGES_INCLUDE, MySqlIncrementalSourceOptions.SCHEMA_CHANGES_EXCLUDE, - 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) + MySqlIncrementalSourceOptions.INT_TYPE_NARROWING) .optional( MySqlIncrementalSourceOptions.STARTUP_MODE, - MySqlIncrementalSourceOptions.STOP_MODE) + Conditions.extension( + MySqlIncrementalSourceOptions.STARTUP_MODE, + MySqlSpecificOffsetGuards.specificStartupModeGuard( + MySqlIncrementalSourceOptions.STARTUP_MODE, + SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE, + SourceOptions.STARTUP_SPECIFIC_OFFSET_POS))) + .optional(MySqlIncrementalSourceOptions.STOP_MODE) .conditional( MySqlIncrementalSourceOptions.STARTUP_MODE, StartupMode.INITIAL, @@ -111,7 +114,47 @@ public class MySqlIncrementalSourceFactory extends BaseChangeStreamTableSourceFa .conditional( MySqlIncrementalSourceOptions.STARTUP_MODE, StartupMode.TIMESTAMP, - SourceOptions.STARTUP_TIMESTAMP); + SourceOptions.STARTUP_TIMESTAMP) + .optional( + SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE, + Conditions.extension( + SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE, + MySqlSpecificOffsetGuards.specificOffsetFileGuard( + MySqlIncrementalSourceOptions.STARTUP_MODE, + SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE, + SourceOptions.STARTUP_SPECIFIC_OFFSET_POS))) + .optional( + SourceOptions.STARTUP_SPECIFIC_OFFSET_POS, + Conditions.extension( + SourceOptions.STARTUP_SPECIFIC_OFFSET_POS, + MySqlSpecificOffsetGuards.specificOffsetPosGuard( + MySqlIncrementalSourceOptions.STARTUP_MODE, + SourceOptions.STARTUP_SPECIFIC_OFFSET_POS, + SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE))) + .optional( + MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_GTID_SET, + Conditions.extension( + MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_GTID_SET, + MySqlSpecificOffsetGuards.specificOffsetGtidSetGuard( + MySqlIncrementalSourceOptions.STARTUP_MODE, + MySqlIncrementalSourceOptions + .STARTUP_SPECIFIC_OFFSET_GTID_SET))) + .optional( + MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_EVENTS, + Conditions.extension( + MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_EVENTS, + MySqlSpecificOffsetGuards.specificOffsetSkipGuard( + MySqlIncrementalSourceOptions.STARTUP_MODE, + MySqlIncrementalSourceOptions + .STARTUP_SPECIFIC_OFFSET_SKIP_EVENTS))) + .optional( + MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_ROWS, + Conditions.extension( + MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_ROWS, + MySqlSpecificOffsetGuards.specificOffsetSkipGuard( + MySqlIncrementalSourceOptions.STARTUP_MODE, + MySqlIncrementalSourceOptions + .STARTUP_SPECIFIC_OFFSET_SKIP_ROWS))); } @Override diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactoryTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactoryTest.java index 37ab496fb0..d55b42aa19 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactoryTest.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactoryTest.java @@ -17,12 +17,215 @@ 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.api.configuration.util.OptionRule; +import org.apache.seatunnel.api.configuration.util.OptionValidationException; +import org.apache.seatunnel.connectors.seatunnel.cdc.mysql.config.MySqlIncrementalSourceOptions; + import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.seatunnel.connectors.cdc.base.option.SourceOptions.STARTUP_SPECIFIC_OFFSET_FILE; +import static org.apache.seatunnel.connectors.cdc.base.option.SourceOptions.STARTUP_SPECIFIC_OFFSET_POS; + public class MySqlIncrementalSourceFactoryTest { + + private static final List<String> SPECIFIC_OFFSET_OPTION_KEYS = + Arrays.asList( + STARTUP_SPECIFIC_OFFSET_FILE.key(), + STARTUP_SPECIFIC_OFFSET_POS.key(), + MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_GTID_SET.key(), + MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_EVENTS.key(), + MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_ROWS.key()); + + private static OptionRule optionRule() { + return new MySqlIncrementalSourceFactory().optionRule(); + } + @Test public void testOptionRule() { Assertions.assertNotNull((new MySqlIncrementalSourceFactory()).optionRule()); } + + @Test + public void testFileAndPosOnlySpecificOffsetsRemainValid() { + Map<String, Object> options = basicOptions(); + options.put(MySqlIncrementalSourceOptions.STARTUP_MODE.key(), "SPECIFIC"); + options.put(STARTUP_SPECIFIC_OFFSET_FILE.key(), "mysql-bin.000004"); + options.put(STARTUP_SPECIFIC_OFFSET_POS.key(), 8937L); + assertValid(options); + } + + @Test + public void testOptionalSpecificOffsetMetadataRemainsValid() { + Map<String, Object> options = basicOptions(); + options.put(MySqlIncrementalSourceOptions.STARTUP_MODE.key(), "SPECIFIC"); + options.put(STARTUP_SPECIFIC_OFFSET_FILE.key(), "mysql-bin.000004"); + options.put(STARTUP_SPECIFIC_OFFSET_POS.key(), 8937L); + options.put( + MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_GTID_SET.key(), + "3E11FA47-71CA-11E1-9E33-C80AA9429562:1-5"); + options.put(MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_EVENTS.key(), 0L); + options.put(MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_ROWS.key(), 0L); + assertValid(options); + } + + @Test + public void testConfigurationWithoutSpecificOffsetsRemainsValid() { + assertValid(basicOptions()); + } + + @Test + public void testExplicitWrongModeRejectsEverySpecificOffset() { + for (String optionKey : SPECIFIC_OFFSET_OPTION_KEYS) { + Map<String, Object> options = basicOptions(); + options.put(MySqlIncrementalSourceOptions.STARTUP_MODE.key(), "EARLIEST"); + options.put(optionKey, sampleValue(optionKey)); + OptionValidationException error = assertValidationFails(options); + Assertions.assertTrue( + error.getMessage().contains("specific-offset") + && error.getMessage().contains("can only be used when"), + () -> + "explicit wrong mode must reject '" + + optionKey + + "': " + + error.getMessage()); + } + } + + @Test + public void testOmittedModeRejectsEverySpecificOffset() { + for (String optionKey : SPECIFIC_OFFSET_OPTION_KEYS) { + Map<String, Object> options = basicOptions(); + options.put(optionKey, sampleValue(optionKey)); + OptionValidationException error = assertValidationFails(options); + Assertions.assertTrue( + error.getMessage().contains("specific-offset") + && error.getMessage().contains("can only be used when"), + () -> "omitted mode must reject '" + optionKey + "': " + error.getMessage()); + } + } + + @Test + public void testSpecificModeWithoutAnchorsIsRejected() { + Map<String, Object> options = basicOptions(); + options.put(MySqlIncrementalSourceOptions.STARTUP_MODE.key(), "SPECIFIC"); + + OptionValidationException error = assertValidationFails(options); + Assertions.assertTrue( + error.getMessage().contains(STARTUP_SPECIFIC_OFFSET_FILE.key()) + && error.getMessage().contains(STARTUP_SPECIFIC_OFFSET_POS.key()) + && error.getMessage() + .contains( + MySqlIncrementalSourceOptions.STARTUP_MODE.key() + + "' requires"), + () -> "specific without anchors must be rejected: " + error.getMessage()); + } + + @Test + public void testPartialAnchorsAreRejected() { + Map<String, Object> fileOnly = basicOptions(); + fileOnly.put(MySqlIncrementalSourceOptions.STARTUP_MODE.key(), "SPECIFIC"); + fileOnly.put(STARTUP_SPECIFIC_OFFSET_FILE.key(), "mysql-bin.000004"); + OptionValidationException fileOnlyError = assertValidationFails(fileOnly); + Assertions.assertTrue( + fileOnlyError.getMessage().contains(STARTUP_SPECIFIC_OFFSET_POS.key()), + () -> "file-only must complain about missing pos: " + fileOnlyError.getMessage()); + + Map<String, Object> posOnly = basicOptions(); + posOnly.put(MySqlIncrementalSourceOptions.STARTUP_MODE.key(), "SPECIFIC"); + posOnly.put(STARTUP_SPECIFIC_OFFSET_POS.key(), 8937L); + OptionValidationException posOnlyError = assertValidationFails(posOnly); + Assertions.assertTrue( + posOnlyError.getMessage().contains(STARTUP_SPECIFIC_OFFSET_FILE.key()), + () -> "pos-only must complain about missing file: " + posOnlyError.getMessage()); + } + + @Test + public void testBlankSpecificOffsetValuesAreRejected() { + Map<String, Object> blankGtid = specificOffsets(); + blankGtid.put(MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_GTID_SET.key(), " "); + OptionValidationException gtidError = assertValidationFails(blankGtid); + Assertions.assertTrue( + gtidError + .getMessage() + .contains( + MySqlIncrementalSourceOptions + .STARTUP_SPECIFIC_OFFSET_GTID_SET + .key()) + && gtidError.getMessage().contains("must not be blank"), + () -> "blank gtid-set must be rejected: " + gtidError.getMessage()); + + Map<String, Object> blankFile = specificOffsets(); + blankFile.put(STARTUP_SPECIFIC_OFFSET_FILE.key(), ""); + OptionValidationException fileError = assertValidationFails(blankFile); + Assertions.assertTrue( + fileError.getMessage().contains(STARTUP_SPECIFIC_OFFSET_FILE.key()) + && fileError.getMessage().contains("must not be blank"), + () -> "blank file must be rejected: " + fileError.getMessage()); + } + + @Test + public void testNegativeSkipValuesAreRejected() { + for (String optionKey : + Arrays.asList( + MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_EVENTS.key(), + MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_ROWS.key())) { + Map<String, Object> options = specificOffsets(); + options.put(optionKey, -1L); + OptionValidationException error = assertValidationFails(options); + Assertions.assertTrue( + error.getMessage().contains(optionKey) + && error.getMessage().contains("must be greater than or equal to 0"), + () -> "negative skip must be rejected: " + error.getMessage()); + } + } + + private static Map<String, Object> basicOptions() { + Map<String, Object> options = new HashMap<>(); + options.put(MySqlIncrementalSourceOptions.USERNAME.key(), "root"); + options.put(MySqlIncrementalSourceOptions.PASSWORD.key(), "password"); + options.put(MySqlIncrementalSourceOptions.URL.key(), "jdbc:mysql://localhost:3306"); + options.put(MySqlIncrementalSourceOptions.TABLE_PATTERN.key(), "db\\..*"); + return options; + } + + private static Map<String, Object> specificOffsets() { + Map<String, Object> options = basicOptions(); + options.put(MySqlIncrementalSourceOptions.STARTUP_MODE.key(), "SPECIFIC"); + options.put(STARTUP_SPECIFIC_OFFSET_FILE.key(), "mysql-bin.000004"); + options.put(STARTUP_SPECIFIC_OFFSET_POS.key(), 8937L); + return options; + } + + private static Object sampleValue(String optionKey) { + if (MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_EVENTS + .key() + .equals(optionKey) + || MySqlIncrementalSourceOptions.STARTUP_SPECIFIC_OFFSET_SKIP_ROWS + .key() + .equals(optionKey)) { + return 0L; + } + if (STARTUP_SPECIFIC_OFFSET_POS.key().equals(optionKey)) { + return 8937L; + } + return "mysql-bin.000004"; + } + + private static void assertValid(Map<String, Object> options) { + ConfigValidator.of(ReadonlyConfig.fromMap(options)).validate(optionRule()); + } + + private static OptionValidationException assertValidationFails(Map<String, Object> options) { + return Assertions.assertThrows( + OptionValidationException.class, + () -> ConfigValidator.of(ReadonlyConfig.fromMap(options)).validate(optionRule())); + } } diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceFactoryTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceFactoryTest.java index 9878a05024..cdf6bb91d0 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceFactoryTest.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oceanbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oceanbase/source/OceanBaseIncrementalSourceFactoryTest.java @@ -98,6 +98,51 @@ public class OceanBaseIncrementalSourceFactoryTest { .validate(new OceanBaseIncrementalSourceFactory().optionRule())); } + /** Verify the inherited specific-offset guards also accept file+pos-only anchors. */ + @Test + public void testOptionRuleAcceptsInheritedFileAndPosOnlySpecificOffsets() { + OceanBaseIncrementalSourceFactory factory = new OceanBaseIncrementalSourceFactory(); + Map<String, Object> options = requiredOptions(); + options.put("startup.mode", "SPECIFIC"); + options.put("startup.specific-offset.file", "ob-bin.000004"); + options.put("startup.specific-offset.pos", 8937L); + + ConfigValidator.of(ReadonlyConfig.fromMap(options)).validate(factory.optionRule()); + } + + /** Verify the inherited guards reject specific offsets when startup.mode points elsewhere. */ + @Test + public void testOptionRuleRejectsInheritedExplicitWrongModeSpecificOffsets() { + Map<String, Object> options = requiredOptions(); + options.put("startup.mode", "EARLIEST"); + options.put("startup.specific-offset.file", "ob-bin.000004"); + + OptionValidationException error = + Assertions.assertThrows( + OptionValidationException.class, + () -> + ConfigValidator.of(ReadonlyConfig.fromMap(options)) + .validate( + new OceanBaseIncrementalSourceFactory() + .optionRule())); + Assertions.assertTrue( + error.getMessage().contains("startup.specific-offset.file"), + () -> "wrong-mode guard must run for OceanBase: " + error.getMessage()); + } + + /** Verify the inherited guards reject specific offsets when startup.mode is omitted. */ + @Test + public void testOptionRuleRejectsInheritedOmittedModeSpecificOffsets() { + Map<String, Object> options = requiredOptions(); + options.put("startup.specific-offset.file", "ob-bin.000004"); + + Assertions.assertThrows( + OptionValidationException.class, + () -> + ConfigValidator.of(ReadonlyConfig.fromMap(options)) + .validate(new OceanBaseIncrementalSourceFactory().optionRule())); + } + private static Map<String, Object> requiredOptions() { Map<String, Object> options = new HashMap<>(); options.put("username", "root");
