This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 b0f6feddf9 [Improve][Connector-V2][MongoDB] Migrate CDC source option
validation to declarative OptionRule (#12735)
b0f6feddf9 is described below
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()));
+ }
}
