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-12735-b3898b62d79db47d4b213c830da8745edd8fc678 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit b0f6feddf9ed0dc1f5d485cfc69d387edfe6ac82 Author: Ryan Yang <[email protected]> AuthorDate: Tue Oct 13 03:07:05 2026 +0000 [Improve][Connector-V2][MongoDB] Migrate CDC source option validation to declarative OptionRule (#12735) --- .../mongodb/MongodbIncrementalSourceFactory.java | 37 ++++- .../config/MongodbTimestampRangeCondition.java | 82 +++++++++ .../MongodbIncrementalSourceFactoryTest.java | 185 +++++++++++++++++++++ 3 files changed, 301 insertions(+), 3 deletions(-) diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/MongodbIncrementalSourceFactory.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/MongodbIncrementalSourceFactory.java index 1aeb1e9071..f22c5b8ae8 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/MongodbIncrementalSourceFactory.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/MongodbIncrementalSourceFactory.java @@ -18,6 +18,7 @@ package org.apache.seatunnel.connectors.seatunnel.cdc.mongodb; 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.options.ConnectorCommonOptions; import org.apache.seatunnel.api.source.SeaTunnelSource; @@ -35,6 +36,7 @@ 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.seatunnel.cdc.mongodb.config.MongodbIncrementalSourceOptions; +import org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbTimestampRangeCondition; import org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.exception.MongodbConnectorException; import com.google.auto.service.AutoService; @@ -68,14 +70,38 @@ public class MongodbIncrementalSourceFactory implements TableSourceFactory { MongodbIncrementalSourceOptions.USERNAME, MongodbIncrementalSourceOptions.PASSWORD, MongodbIncrementalSourceOptions.CONNECTION_OPTIONS, + MongodbIncrementalSourceOptions.STARTUP_MODE, + MongodbIncrementalSourceOptions.STOP_MODE, + MongodbIncrementalSourceOptions.DEBEZIUM_PROPERTIES) + // The value domains below mirror the checks the source applies while the job is + // built, so an invalid value fails the validation of the job configuration instead + // of the job startup. + .optional( + MongodbIncrementalSourceOptions.SCHEMA, + Conditions.mapNotEmpty(MongodbIncrementalSourceOptions.SCHEMA)) + .optional( + MongodbIncrementalSourceOptions.TABLE_CONFIGS, + Conditions.notEmpty(MongodbIncrementalSourceOptions.TABLE_CONFIGS)) + .optional( MongodbIncrementalSourceOptions.BATCH_SIZE, + Conditions.greaterOrEqual(MongodbIncrementalSourceOptions.BATCH_SIZE, 0)) + .optional( MongodbIncrementalSourceOptions.POLL_MAX_BATCH_SIZE, + Conditions.greaterThan( + MongodbIncrementalSourceOptions.POLL_MAX_BATCH_SIZE, 0)) + .optional( MongodbIncrementalSourceOptions.POLL_AWAIT_TIME_MILLIS, + Conditions.greaterThan( + MongodbIncrementalSourceOptions.POLL_AWAIT_TIME_MILLIS, 0)) + .optional( MongodbIncrementalSourceOptions.HEARTBEAT_INTERVAL_MILLIS, + Conditions.greaterOrEqual( + MongodbIncrementalSourceOptions.HEARTBEAT_INTERVAL_MILLIS, 0)) + .optional( MongodbIncrementalSourceOptions.INCREMENTAL_SNAPSHOT_CHUNK_SIZE_MB, - MongodbIncrementalSourceOptions.STARTUP_MODE, - MongodbIncrementalSourceOptions.STOP_MODE, - MongodbIncrementalSourceOptions.DEBEZIUM_PROPERTIES) + Conditions.greaterThan( + MongodbIncrementalSourceOptions.INCREMENTAL_SNAPSHOT_CHUNK_SIZE_MB, + 0)) .conditional( MongodbIncrementalSourceOptions.STARTUP_MODE, StartupMode.TIMESTAMP, @@ -84,6 +110,11 @@ public class MongodbIncrementalSourceFactory implements TableSourceFactory { MongodbIncrementalSourceOptions.STOP_MODE, StopMode.TIMESTAMP, SourceOptions.STOP_TIMESTAMP) + .conditional( + MongodbIncrementalSourceOptions.STOP_MODE, + StopMode.TIMESTAMP, + Conditions.extension( + SourceOptions.STOP_TIMESTAMP, new MongodbTimestampRangeCondition())) .build(); } diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/config/MongodbTimestampRangeCondition.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/config/MongodbTimestampRangeCondition.java new file mode 100644 index 0000000000..9ecd269c13 --- /dev/null +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/config/MongodbTimestampRangeCondition.java @@ -0,0 +1,82 @@ +/* + * 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.mongodb.config; + +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.config.StartupConfig; +import org.apache.seatunnel.connectors.cdc.base.config.StopConfig; +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.source.offset.Offset; +import org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.source.offset.ChangeStreamOffsetFactory; + +/** + * Rejects a {@code stop.timestamp} that does not resolve to a change-stream position after the + * configured {@code startup.timestamp}. + * + * <p>The two millisecond values are compared as the positions they resolve to, because MongoDB + * change-stream timestamps only carry second precision: two different millisecond values inside the + * same second resolve to the same position and are rejected as well. The check only applies when + * the startup mode is {@code timestamp}; the stop mode is {@code timestamp} whenever the condition + * is evaluated. + */ +public class MongodbTimestampRangeCondition implements ConditionExtension<Long> { + + @Override + public String description() { + return "must resolve to a change-stream position after '" + + SourceOptions.STARTUP_TIMESTAMP.key() + + "' when '" + + SourceOptions.STARTUP_MODE_KEY + + "' is 'timestamp'"; + } + + @Override + public boolean evaluate(ReadonlyConfig config, Long stopTimestamp) + throws OptionValidationException { + StartupMode startupMode = + config.getOptional(MongodbIncrementalSourceOptions.STARTUP_MODE) + .orElseGet(MongodbIncrementalSourceOptions.STARTUP_MODE::defaultValue); + Long startupTimestamp = config.getOptional(SourceOptions.STARTUP_TIMESTAMP).orElse(null); + if (stopTimestamp == null + || startupTimestamp == null + || startupMode != StartupMode.TIMESTAMP) { + return true; + } + ChangeStreamOffsetFactory offsetFactory = new ChangeStreamOffsetFactory(); + Offset startupOffset = + new StartupConfig(StartupMode.TIMESTAMP, null, null, startupTimestamp) + .getStartupOffset(offsetFactory); + Offset stopOffset = + new StopConfig(StopMode.TIMESTAMP, null, null, stopTimestamp) + .getStopOffset(offsetFactory); + if (stopOffset.compareTo(startupOffset) <= 0) { + throw new OptionValidationException( + String.format( + "Invalid timestamp range: stop.timestamp (%d) resolves to MongoDB " + + "position %s, which must be after startup.timestamp (%d) at " + + "position %s. MongoDB change-stream timestamps have second " + + "precision.", + stopTimestamp, stopOffset, startupTimestamp, startupOffset)); + } + return true; + } +} diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/mongodb/source/MongodbIncrementalSourceFactoryTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/mongodb/source/MongodbIncrementalSourceFactoryTest.java index f40f4194f3..300752d933 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/mongodb/source/MongodbIncrementalSourceFactoryTest.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/mongodb/source/MongodbIncrementalSourceFactoryTest.java @@ -17,7 +17,12 @@ package mongodb.source; +import org.apache.seatunnel.api.configuration.Option; +import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.api.configuration.SingleChoiceOption; +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.cdc.base.config.StartupConfig; import org.apache.seatunnel.connectors.cdc.base.config.StopConfig; import org.apache.seatunnel.connectors.cdc.base.option.SourceOptions; @@ -25,6 +30,7 @@ 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.source.offset.Offset; import org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.MongodbIncrementalSourceFactory; +import org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbIncrementalSourceOptions; import org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConfig; import org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConfigProvider; import org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.exception.MongodbConnectorException; @@ -34,7 +40,11 @@ import org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.source.offset.Chang import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; public class MongodbIncrementalSourceFactoryTest { @Test @@ -218,4 +228,179 @@ public class MongodbIncrementalSourceFactoryTest { MongodbSourceConfigProvider.newBuilder() .stopOptions(new StopConfig(StopMode.LATEST, null, null, null))); } + + @Test + public void testOptionRuleAcceptsValidValueDomains() { + assertValid(basicOptions()); + } + + @Test + public void testOptionRuleAcceptsBoundaryNumericOptions() { + Map<String, Object> options = basicOptions(); + options.put(MongodbIncrementalSourceOptions.BATCH_SIZE.key(), 0); + options.put(MongodbIncrementalSourceOptions.POLL_MAX_BATCH_SIZE.key(), 1); + options.put(MongodbIncrementalSourceOptions.POLL_AWAIT_TIME_MILLIS.key(), 1); + options.put(MongodbIncrementalSourceOptions.HEARTBEAT_INTERVAL_MILLIS.key(), 0); + options.put(MongodbIncrementalSourceOptions.INCREMENTAL_SNAPSHOT_CHUNK_SIZE_MB.key(), 1); + assertValid(options); + } + + @Test + public void testOptionRuleRejectsOutOfRangeNumericOptions() { + Object[][] invalidValues = { + {MongodbIncrementalSourceOptions.BATCH_SIZE, -1, ">= 0"}, + {MongodbIncrementalSourceOptions.POLL_MAX_BATCH_SIZE, 0, "> 0"}, + {MongodbIncrementalSourceOptions.POLL_AWAIT_TIME_MILLIS, 0, "> 0"}, + {MongodbIncrementalSourceOptions.HEARTBEAT_INTERVAL_MILLIS, -1, ">= 0"}, + {MongodbIncrementalSourceOptions.INCREMENTAL_SNAPSHOT_CHUNK_SIZE_MB, 0, "> 0"}, + }; + for (Object[] invalidValue : invalidValues) { + Option<?> option = (Option<?>) invalidValue[0]; + String operator = (String) invalidValue[2]; + Map<String, Object> options = basicOptions(); + options.put(option.key(), invalidValue[1]); + OptionValidationException error = assertValidationFails(options); + Assertions.assertTrue( + error.getMessage().contains(option.key()) + && error.getMessage().contains(operator), + () -> + option.key() + + " must be rejected by '" + + operator + + "': " + + error.getMessage()); + } + } + + @Test + public void testOptionRuleAcceptsTableConfigsInsteadOfSchema() { + Map<String, Object> options = basicOptions(); + options.remove(MongodbIncrementalSourceOptions.SCHEMA.key()); + options.put( + MongodbIncrementalSourceOptions.TABLE_CONFIGS.key(), + Collections.singletonList(schemaEntry())); + assertValid(options); + } + + @Test + public void testOptionRuleRejectsEmptySchema() { + Map<String, Object> options = basicOptions(); + options.put(MongodbIncrementalSourceOptions.SCHEMA.key(), new HashMap<String, Object>()); + OptionValidationException error = assertValidationFails(options); + Assertions.assertTrue(error.getMessage().contains("is not empty"), error.getMessage()); + } + + @Test + public void testOptionRuleRejectsEmptyTableConfigs() { + Map<String, Object> options = basicOptions(); + options.remove(MongodbIncrementalSourceOptions.SCHEMA.key()); + options.put( + MongodbIncrementalSourceOptions.TABLE_CONFIGS.key(), + new ArrayList<Map<String, Object>>()); + OptionValidationException error = assertValidationFails(options); + Assertions.assertTrue(error.getMessage().contains("is not empty"), error.getMessage()); + } + + @Test + public void testOptionRuleKeepsSchemaAndTableConfigsExclusive() { + Map<String, Object> options = basicOptions(); + options.put( + MongodbIncrementalSourceOptions.TABLE_CONFIGS.key(), + Collections.singletonList(schemaEntry())); + OptionValidationException error = assertValidationFails(options); + Assertions.assertTrue( + error.getMessage().contains("mutually exclusive"), error.getMessage()); + } + + @Test + public void testOptionRuleRejectsMissingSchemaAndTableConfigs() { + Map<String, Object> options = basicOptions(); + options.remove(MongodbIncrementalSourceOptions.SCHEMA.key()); + OptionValidationException error = assertValidationFails(options); + Assertions.assertTrue( + error.getMessage().contains("exactly one option must be set"), error.getMessage()); + } + + @Test + public void testOptionRuleRequiresStopTimestampWhenStopModeIsTimestamp() { + // stop.timestamp is required when stop.mode is timestamp; the range check added along with + // it must not shadow that requirement. + Map<String, Object> options = basicOptions(); + options.put(MongodbIncrementalSourceOptions.STOP_MODE.key(), "timestamp"); + OptionValidationException error = assertValidationFails(options); + Assertions.assertTrue( + error.getMessage().contains(SourceOptions.STOP_TIMESTAMP.key()), + error.getMessage()); + } + + @Test + public void testOptionRuleAcceptsTimestampRange() { + assertValid(timestampRangeOptions(1_700_000_000_000L, 1_700_000_001_000L)); + } + + @Test + public void testOptionRuleRejectsTimestampRangeThatDoesNotAdvance() { + long startupTimestamp = 1_700_000_000_000L; + + // Comparing raw milliseconds would accept both of these; the positions they resolve to are + // what matters, and a change-stream timestamp only carries second precision. + OptionValidationException sameSecond = + assertValidationFails( + timestampRangeOptions(startupTimestamp, startupTimestamp + 500L)); + Assertions.assertTrue( + sameSecond.getMessage().contains("second precision"), sameSecond.getMessage()); + assertValidationFails(timestampRangeOptions(startupTimestamp, startupTimestamp)); + assertValidationFails(timestampRangeOptions(startupTimestamp, startupTimestamp - 1000L)); + } + + @Test + public void testOptionRuleIgnoresTimestampRangeWhenStartupModeIsNotTimestamp() { + Map<String, Object> options = timestampRangeOptions(1_700_000_000_000L, 1_700_000_000_000L); + options.put(MongodbIncrementalSourceOptions.STARTUP_MODE.key(), "latest"); + options.remove(SourceOptions.STARTUP_TIMESTAMP.key()); + assertValid(options); + } + + private static OptionRule optionRule() { + return new MongodbIncrementalSourceFactory().optionRule(); + } + + private static Map<String, Object> basicOptions() { + Map<String, Object> options = new HashMap<>(); + options.put(MongodbIncrementalSourceOptions.HOSTS.key(), "localhost:27017"); + options.put( + MongodbIncrementalSourceOptions.DATABASE.key(), + Collections.singletonList("inventory")); + options.put( + MongodbIncrementalSourceOptions.COLLECTION.key(), + Collections.singletonList("products")); + options.put(MongodbIncrementalSourceOptions.SCHEMA.key(), schemaEntry()); + return options; + } + + private static Map<String, Object> schemaEntry() { + Map<String, Object> table = new HashMap<>(); + table.put("fields", Collections.singletonMap("id", "string")); + return Collections.singletonMap("inventory.products", table); + } + + private static Map<String, Object> timestampRangeOptions( + long startupTimestamp, long stopTimestamp) { + Map<String, Object> options = basicOptions(); + options.put(MongodbIncrementalSourceOptions.STARTUP_MODE.key(), "timestamp"); + options.put(SourceOptions.STARTUP_TIMESTAMP.key(), startupTimestamp); + options.put(MongodbIncrementalSourceOptions.STOP_MODE.key(), "timestamp"); + options.put(SourceOptions.STOP_TIMESTAMP.key(), stopTimestamp); + return options; + } + + 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())); + } }
