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 30fb930522 [Improve][Connector-V2] Migrate Kafka Source imperative
validation to declarative OptionRule (#11157)
30fb930522 is described below
commit 30fb9305220eea28d2e584a5904150a97663ae23
Author: zhiwei.niu <[email protected]>
AuthorDate: Wed Jul 8 22:41:52 2026 +0800
[Improve][Connector-V2] Migrate Kafka Source imperative validation to
declarative OptionRule (#11157)
---
.../seatunnel/kafka/source/KafkaSourceConfig.java | 28 +--
.../seatunnel/kafka/source/KafkaSourceFactory.java | 188 ++++++++++++++++--
.../seatunnel/kafka/KafkaFactoryTest.java | 220 +++++++++++++++++++++
.../seatunnel/e2e/connector/kafka/KafkaIT.java | 121 +++++++++++-
...bles_configs_start_mode_earliest_to_assert.conf | 64 ++++++
...configs_start_mode_group_offsets_to_assert.conf | 65 ++++++
...tables_configs_start_mode_latest_to_assert.conf | 56 ++++++
...figs_start_mode_specific_offsets_to_assert.conf | 76 +++++++
...les_configs_start_mode_timestamp_to_assert.conf | 65 ++++++
9 files changed, 847 insertions(+), 36 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java
index 0913f47ebf..bd689ecea4 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java
@@ -41,7 +41,6 @@ import
org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
import
org.apache.seatunnel.connectors.seatunnel.kafka.config.KafkaSourceOptions;
import org.apache.seatunnel.connectors.seatunnel.kafka.config.MessageFormat;
import
org.apache.seatunnel.connectors.seatunnel.kafka.config.MessageFormatErrorHandleWay;
-import org.apache.seatunnel.connectors.seatunnel.kafka.config.StartMode;
import
org.apache.seatunnel.connectors.seatunnel.kafka.config.TableIdentifierConfig;
import org.apache.seatunnel.format.avro.AvroDeserializationSchema;
import
org.apache.seatunnel.format.compatible.kafka.connect.json.CompatibleKafkaConnectDeserializationSchema;
@@ -59,7 +58,6 @@ import
org.apache.seatunnel.format.protobuf.SchemaRegistryAwareProtobufDeseriali
import org.apache.seatunnel.format.text.TextDeserializationSchema;
import org.apache.seatunnel.format.text.constant.TextFormatConstant;
-import org.apache.commons.collections4.MapUtils;
import org.apache.kafka.common.TopicPartition;
import lombok.Getter;
@@ -132,11 +130,6 @@ public class KafkaSourceConfig implements Serializable {
this.consumerGroup = readonlyConfig.get(CONSUMER_GROUP);
this.readerCacheQueueSize =
readonlyConfig.get(READER_CACHE_QUEUE_SIZE);
this.ignoreNoLeaderPartition =
readonlyConfig.get(IGNORE_NO_LEADER_PARTITION);
- if (this.ignoreNoLeaderPartition && this.discoveryIntervalMillis <= 0)
{
- throw new IllegalArgumentException(
- "partition-discovery.interval-millis must be configured
when ignore_no_leader_partition is set to true. "
- + "Please provide a positive value for
partition-discovery.interval-millis.");
- }
}
private Properties createKafkaProperties(ReadonlyConfig readonlyConfig) {
@@ -196,10 +189,11 @@ public class KafkaSourceConfig implements Serializable {
long startOffsetsTimestamp =
readonlyConfig.get(START_MODE_TIMESTAMP);
long currentTimestamp =
System.currentTimeMillis();
- if (startOffsetsTimestamp < 0
- || startOffsetsTimestamp >
currentTimestamp) {
+ // Runtime check: cannot be declarative
(depends on current
+ // time)
+ if (startOffsetsTimestamp >
currentTimestamp) {
throw new IllegalArgumentException(
- "start_mode.timestamp The
value is smaller than 0 or smaller than the current time");
+ "start_mode.timestamp must not
be greater than the current time");
}
consumerMetadata.setStartOffsetsTimestamp(
startOffsetsTimestamp);
@@ -207,25 +201,19 @@ public class KafkaSourceConfig implements Serializable {
readonlyConfig.get(START_MODE_END_TIMESTAMP))) {
long endOffsetsTimestamp =
readonlyConfig.get(START_MODE_END_TIMESTAMP);
- if (endOffsetsTimestamp < 0
- || endOffsetsTimestamp >
currentTimestamp) {
+ // Runtime check: cannot be
declarative (depends on current
+ // time)
+ if (endOffsetsTimestamp >
currentTimestamp) {
throw new IllegalArgumentException(
- "start_mode.endTimestamp
The value is smaller than 0 or smaller than the current time");
+ "start_mode.end_timestamp
must not be greater than the current time");
}
consumerMetadata.setEndOffsetsTimestamp(
endOffsetsTimestamp);
}
break;
case SPECIFIC_OFFSETS:
- // Key is topic-partition, value is offset
Map<String, Long> offsetMap =
readonlyConfig.get(START_MODE_OFFSETS);
- if (MapUtils.isEmpty(offsetMap)) {
- throw new IllegalArgumentException(
- "start mode is "
- +
StartMode.SPECIFIC_OFFSETS
- + "but no specific
offsets were specified.");
- }
Map<TopicPartition, Long>
specificStartOffsets =
new HashMap<>();
offsetMap.forEach(
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java
index 4751d7982a..6e54ea69fa 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java
@@ -17,7 +17,11 @@
package org.apache.seatunnel.connectors.seatunnel.kafka.source;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
+import org.apache.seatunnel.api.configuration.util.Conditions;
import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
import org.apache.seatunnel.api.source.SeaTunnelSource;
import org.apache.seatunnel.api.source.SourceSplit;
import org.apache.seatunnel.api.table.connector.TableSource;
@@ -31,6 +35,8 @@ import
org.apache.seatunnel.connectors.seatunnel.kafka.config.StartMode;
import com.google.auto.service.AutoService;
import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
@AutoService(Factory.class)
public class KafkaSourceFactory implements TableSourceFactory {
@@ -49,7 +55,6 @@ public class KafkaSourceFactory implements TableSourceFactory
{
KafkaSourceOptions.TABLE_CONFIGS,
KafkaSourceOptions.TABLE_LIST)
.optional(
- KafkaSourceOptions.START_MODE,
KafkaSourceOptions.PATTERN,
KafkaSourceOptions.CONSUMER_GROUP,
KafkaSourceOptions.COMMIT_ON_CHECKPOINT,
@@ -60,24 +65,41 @@ public class KafkaSourceFactory implements
TableSourceFactory {
KafkaSourceOptions.DEBEZIUM_RECORD_INCLUDE_SCHEMA,
KafkaSourceOptions.DEBEZIUM_RECORD_TABLE_FILTER,
KafkaSourceOptions.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS,
- KafkaSourceOptions.READER_CACHE_QUEUE_SIZE,
- KafkaSourceOptions.IGNORE_NO_LEADER_PARTITION)
- .conditional(
- KafkaSourceOptions.START_MODE,
- StartMode.TIMESTAMP,
- KafkaSourceOptions.START_MODE_TIMESTAMP)
- .conditional(
- KafkaSourceOptions.IGNORE_NO_LEADER_PARTITION,
- Boolean.TRUE,
-
KafkaSourceOptions.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS)
- .conditional(
- KafkaSourceOptions.START_MODE,
- StartMode.SPECIFIC_OFFSETS,
- KafkaSourceOptions.START_MODE_OFFSETS)
+ KafkaSourceOptions.READER_CACHE_QUEUE_SIZE)
+ .optional(
+ KafkaSourceOptions.START_MODE_TIMESTAMP,
+ Conditions.extension(
+ KafkaSourceOptions.START_MODE_TIMESTAMP,
+ new KafkaStartModeTimestampValidator()))
+ .optional(
+ KafkaSourceOptions.START_MODE_OFFSETS,
+ Conditions.extension(
+ KafkaSourceOptions.START_MODE_OFFSETS,
+ new KafkaStartModeOffsetsValidator()))
+ .optional(
+ KafkaSourceOptions.START_MODE_END_TIMESTAMP,
+
Conditions.greaterOrEqual(KafkaSourceOptions.START_MODE_END_TIMESTAMP, 0L))
+ .optional(
+ KafkaSourceOptions.TABLE_CONFIGS,
+ Conditions.extension(
+ KafkaSourceOptions.TABLE_CONFIGS, new
KafkaTableConfigsValidator()))
+ .optional(
+ KafkaSourceOptions.TABLE_LIST,
+ Conditions.extension(
+ KafkaSourceOptions.TABLE_LIST, new
KafkaTableConfigsValidator()))
.conditional(
KafkaSourceOptions.FORMAT,
MessageFormat.PROTOBUF,
KafkaSourceOptions.STRIP_SCHEMA_REGISTRY_HEADER)
+ .optional(
+ KafkaSourceOptions.START_MODE,
+ Conditions.extension(
+ KafkaSourceOptions.START_MODE, new
KafkaStartModeValidator()))
+ .optional(
+ KafkaSourceOptions.IGNORE_NO_LEADER_PARTITION,
+ Conditions.extension(
+ KafkaSourceOptions.IGNORE_NO_LEADER_PARTITION,
+ new KafkaPartitionDiscoveryValidator()))
.build();
}
@@ -91,4 +113,140 @@ public class KafkaSourceFactory implements
TableSourceFactory {
public Class<? extends SeaTunnelSource> getSourceClass() {
return KafkaSource.class;
}
+
+ private static class KafkaPartitionDiscoveryValidator implements
ConditionExtension<Boolean> {
+ @Override
+ public String description() {
+ return "If [ignore_no_leader_partition] is true, "
+ + "then [partition-discovery.interval-millis] must not be
null and greater than 0";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, Boolean value)
+ throws OptionValidationException {
+ if (!value) {
+ return true;
+ }
+ Long partitionDiscoveryIntervalMillis =
+
config.get(KafkaSourceOptions.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS);
+ return partitionDiscoveryIntervalMillis != null &&
partitionDiscoveryIntervalMillis > 0;
+ }
+ }
+
+ private static class KafkaStartModeValidator implements
ConditionExtension<StartMode> {
+ @Override
+ public String description() {
+ return "if [start_mode] is timestamp then [start_mode.timestamp]
must be configured, "
+ + "if [start_mode] is specific_offsets then
[start_mode.offsets] must be configured";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, StartMode value)
+ throws OptionValidationException {
+ if (StartMode.TIMESTAMP == value) {
+ return
config.getOptional(KafkaSourceOptions.START_MODE_TIMESTAMP).isPresent();
+
+ } else if (StartMode.SPECIFIC_OFFSETS == value) {
+ Map<String, Long> startModeOffsets =
+ config.get(KafkaSourceOptions.START_MODE_OFFSETS);
+ return startModeOffsets != null && !startModeOffsets.isEmpty();
+ }
+ return true;
+ }
+ }
+
+ private static class KafkaStartModeTimestampValidator implements
ConditionExtension<Long> {
+ @Override
+ public String description() {
+ return "[start_mode.timestamp] is only valid when
[start_mode]=timestamp and must be >= 0";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, Long value)
+ throws OptionValidationException {
+ StartMode startMode =
config.getOptional(KafkaSourceOptions.START_MODE).orElse(null);
+ return startMode == StartMode.TIMESTAMP && value != null && value
>= 0L;
+ }
+ }
+
+ private static class KafkaStartModeOffsetsValidator
+ implements ConditionExtension<Map<String, Long>> {
+ @Override
+ public String description() {
+ return "[start_mode.offsets] is only valid when
[start_mode]=specific_offsets and must not be empty";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, Map<String, Long> value)
+ throws OptionValidationException {
+ StartMode startMode =
config.getOptional(KafkaSourceOptions.START_MODE).orElse(null);
+ return startMode == StartMode.SPECIFIC_OFFSETS && value != null &&
!value.isEmpty();
+ }
+ }
+
+ private static class KafkaTableConfigsValidator
+ implements ConditionExtension<List<Map<String, Object>>> {
+
+ @Override
+ public String description() {
+ return "each tables_configs entry: start_mode.timestamp >= 0 when
TIMESTAMP, "
+ + "non-empty start_mode.offsets when SPECIFIC_OFFSETS";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, List<Map<String,
Object>> entries)
+ throws OptionValidationException {
+ if (entries == null || entries.isEmpty()) {
+ return true;
+ }
+ for (int i = 0; i < entries.size(); i++) {
+ ReadonlyConfig entryConfig =
ReadonlyConfig.fromMap(entries.get(i));
+ StartMode startMode =
+
entryConfig.getOptional(KafkaSourceOptions.START_MODE).orElse(null);
+ if (startMode == StartMode.TIMESTAMP) {
+ Long ts =
entryConfig.get(KafkaSourceOptions.START_MODE_TIMESTAMP);
+ if (ts == null || ts < 0) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: 'start_mode.timestamp'
must be >= 0, got: %d",
+ i, ts);
+ }
+ if
(entries.get(i).containsKey(KafkaSourceOptions.START_MODE_OFFSETS.key())) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: 'start_mode.offsets' is
only valid "
+ + "when start_mode=SPECIFIC_OFFSETS",
+ i);
+ }
+ Long endTs =
entryConfig.get(KafkaSourceOptions.START_MODE_END_TIMESTAMP);
+ if (endTs != null && endTs < 0) {
+ throw new OptionValidationException(
+ "tables_configs[%d]:
'start_mode.end_timestamp' must be >= 0, got: %d",
+ i, endTs);
+ }
+ } else if (startMode == StartMode.SPECIFIC_OFFSETS) {
+ Map<String, Long> offsets =
+
entryConfig.get(KafkaSourceOptions.START_MODE_OFFSETS);
+ if (offsets == null || offsets.isEmpty()) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: 'start_mode.offsets' must
not be empty "
+ + "when start_mode=SPECIFIC_OFFSETS",
+ i);
+ }
+ if
(entries.get(i).containsKey(KafkaSourceOptions.START_MODE_TIMESTAMP.key())) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: 'start_mode.timestamp' is
only valid "
+ + "when start_mode=TIMESTAMP",
+ i);
+ }
+ } else if
(entries.get(i).containsKey(KafkaSourceOptions.START_MODE_TIMESTAMP.key())
+ || entries.get(i)
+
.containsKey(KafkaSourceOptions.START_MODE_OFFSETS.key())) {
+ throw new OptionValidationException(
+ "tables_configs[%d]: 'start_mode.timestamp' and "
+ + "'start_mode.offsets' require an
appropriate start_mode",
+ i);
+ }
+ }
+ return true;
+ }
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/KafkaFactoryTest.java
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/KafkaFactoryTest.java
index 7ab57b02f0..6af5fa7937 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/KafkaFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/KafkaFactoryTest.java
@@ -17,17 +17,237 @@
package org.apache.seatunnel.connectors.seatunnel.kafka;
+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.kafka.sink.KafkaSinkFactory;
import
org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceFactory;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
class KafkaFactoryTest {
+ private final OptionRule sourceRule = new
KafkaSourceFactory().optionRule();
+
+ private void validate(Map<String, Object> config) {
+
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(sourceRule);
+ }
+
+ private void validateUnknownKeys(Map<String, Object> config) {
+ ConfigValidator.validateUnknownKeys(
+ ReadonlyConfig.fromMap(config), sourceRule, "KafkaSource");
+ }
+
+ private Map<String, Object> validTimestampConfig() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("bootstrap.servers", "localhost:9092");
+ cfg.put("topic", "test-topic");
+ cfg.put("start_mode", "TIMESTAMP");
+ cfg.put("start_mode.timestamp", 1000L);
+ return cfg;
+ }
+
+ private Map<String, Object> validSpecificOffsetsConfig() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("bootstrap.servers", "localhost:9092");
+ cfg.put("topic", "test-topic");
+ cfg.put("start_mode", "SPECIFIC_OFFSETS");
+ Map<String, Long> offsets = new HashMap<>();
+ offsets.put("test-topic-0", 100L);
+ cfg.put("start_mode.offsets", offsets);
+ return cfg;
+ }
+
@Test
void optionRule() {
Assertions.assertNotNull((new KafkaSourceFactory()).optionRule());
Assertions.assertNotNull((new KafkaSinkFactory()).optionRule());
}
+
+ @Test
+ void testValidTimestampConfig() {
+ Assertions.assertDoesNotThrow(() -> validate(validTimestampConfig()));
+ }
+
+ @Test
+ void testUnknownKeysAllowsTimestampStartModeOptions() {
+ Assertions.assertDoesNotThrow(() ->
validateUnknownKeys(validTimestampConfig()));
+ }
+
+ @Test
+ void testNegativeTimestampRejected() {
+ Map<String, Object> cfg = validTimestampConfig();
+ cfg.put("start_mode.timestamp", -1L);
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(cfg));
+ }
+
+ @Test
+ void testTimestampRejectedWithoutTimestampStartMode() {
+ Map<String, Object> cfg = validTimestampConfig();
+ cfg.put("start_mode", "EARLIEST");
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(cfg));
+ }
+
+ @Test
+ void testTimestampRejectedWithoutStartMode() {
+ Map<String, Object> cfg = validTimestampConfig();
+ cfg.remove("start_mode");
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(cfg));
+ }
+
+ @Test
+ void testNegativeEndTimestampRejected() {
+ Map<String, Object> cfg = validTimestampConfig();
+ cfg.put("start_mode.end_timestamp", -1L);
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(cfg));
+ }
+
+ @Test
+ void testValidSpecificOffsetsConfig() {
+ Assertions.assertDoesNotThrow(() ->
validate(validSpecificOffsetsConfig()));
+ }
+
+ @Test
+ void testUnknownKeysAllowsSpecificOffsetsStartModeOptions() {
+ Assertions.assertDoesNotThrow(() ->
validateUnknownKeys(validSpecificOffsetsConfig()));
+ }
+
+ @Test
+ void testEmptyOffsetsMapRejected() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("bootstrap.servers", "localhost:9092");
+ cfg.put("topic", "test-topic");
+ cfg.put("start_mode", "SPECIFIC_OFFSETS");
+ cfg.put("start_mode.offsets", Collections.emptyMap());
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(cfg));
+ }
+
+ @Test
+ void testOffsetsRejectedWithoutSpecificOffsetsStartMode() {
+ Map<String, Object> cfg = validSpecificOffsetsConfig();
+ cfg.put("start_mode", "EARLIEST");
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(cfg));
+ }
+
+ @Test
+ void testOffsetsRejectedWithoutStartMode() {
+ Map<String, Object> cfg = validSpecificOffsetsConfig();
+ cfg.remove("start_mode");
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(cfg));
+ }
+
+ @Test
+ void testIgnoreNoLeaderPartition() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("bootstrap.servers", "localhost:9092");
+ cfg.put("topic", "test-topic");
+ cfg.put("ignore_no_leader_partition", false);
+ Assertions.assertDoesNotThrow(() -> validate(cfg));
+
+ cfg.put("ignore_no_leader_partition", true);
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(cfg));
+
+ cfg.put("partition-discovery.interval-millis", 0);
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(cfg));
+
+ cfg.put("partition-discovery.interval-millis", 10);
+ Assertions.assertDoesNotThrow(() -> validate(cfg));
+ }
+
+ // --- Multi-table (tables_configs) tests ---
+
+ private Map<String, Object> validMultiTableConfig() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("bootstrap.servers", "localhost:9092");
+
+ Map<String, Object> entry = new HashMap<>();
+ entry.put("topic", "multi-topic");
+ entry.put("start_mode", "TIMESTAMP");
+ entry.put("start_mode.timestamp", 5000L);
+ List<Map<String, Object>> tables = new ArrayList<>();
+ tables.add(entry);
+ cfg.put("tables_configs", tables);
+ return cfg;
+ }
+
+ private Map<String, Object> validTableListConfig() {
+ Map<String, Object> cfg = validMultiTableConfig();
+ cfg.put("table_list", cfg.remove("tables_configs"));
+ return cfg;
+ }
+
+ @Test
+ void testMultiTableValidConfig() {
+ Assertions.assertDoesNotThrow(() -> validate(validMultiTableConfig()));
+ }
+
+ @Test
+ void testTableListValidConfig() {
+ Assertions.assertDoesNotThrow(() -> validate(validTableListConfig()));
+ }
+
+ @Test
+ void testMultiTableNegativeTimestampRejected() {
+ Map<String, Object> cfg = validMultiTableConfig();
+ @SuppressWarnings("unchecked")
+ List<Map<String, Object>> tables = (List<Map<String, Object>>)
cfg.get("tables_configs");
+ tables.get(0).put("start_mode.timestamp", null);
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(cfg));
+ }
+
+ @Test
+ void testMultiTableNegativeEndTimestampRejected() {
+ Map<String, Object> cfg = validMultiTableConfig();
+ @SuppressWarnings("unchecked")
+ List<Map<String, Object>> tables = (List<Map<String, Object>>)
cfg.get("tables_configs");
+ tables.get(0).put("start_mode.end_timestamp", -1L);
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(cfg));
+ }
+
+ @Test
+ void testMultiTableTimestampRejectedWithoutTimestampStartMode() {
+ Map<String, Object> cfg = validMultiTableConfig();
+ @SuppressWarnings("unchecked")
+ List<Map<String, Object>> tables = (List<Map<String, Object>>)
cfg.get("tables_configs");
+ tables.get(0).put("start_mode", "EARLIEST");
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(cfg));
+ }
+
+ @Test
+ void testMultiTableEmptyOffsetsRejected() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("bootstrap.servers", "localhost:9092");
+
+ Map<String, Object> entry = new HashMap<>();
+ entry.put("topic", "multi-topic");
+ entry.put("start_mode", "SPECIFIC_OFFSETS");
+ entry.put("start_mode.offsets", Collections.emptyMap());
+ List<Map<String, Object>> tables = new ArrayList<>();
+ tables.add(entry);
+ cfg.put("tables_configs", tables);
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(cfg));
+ }
+
+ @Test
+ void testMultiTableOffsetsRejectedWithoutSpecificOffsetsStartMode() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("bootstrap.servers", "localhost:9092");
+
+ Map<String, Object> entry = new HashMap<>();
+ entry.put("topic", "multi-topic");
+ entry.put("start_mode", "EARLIEST");
+ entry.put("start_mode.offsets",
Collections.singletonMap("multi-topic-0", 1L));
+ List<Map<String, Object>> tables = new ArrayList<>();
+ tables.add(entry);
+ cfg.put("tables_configs", tables);
+ Assertions.assertThrows(OptionValidationException.class, () ->
validate(cfg));
+ }
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
index 72ac59dc22..751031dca6 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
@@ -106,12 +106,14 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
@@ -143,6 +145,8 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
/** Topics created dynamically during tests; cleaned up in {@link
#tearDown()}. */
private final List<String> dynamicTopics = new CopyOnWriteArrayList<>();
+ private final AtomicInteger startModeTestSequence = new AtomicInteger();
+
@BeforeAll
@Override
public void startUp() throws Exception {
@@ -1119,6 +1123,96 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
}
+ @TestTemplate
+ public void testSourceKafkaTablesConfigsEarliestStartMode(TestContainer
container)
+ throws IOException, InterruptedException {
+ String topic = startModeTopic(container, "earliest");
+ createKafkaTopic(topic);
+ generateSimpleJsonRecords(topic, 0, 10);
+
+ Container.ExecResult execResult =
+ container.executeJob(
+
"/kafka/kafkasource_tables_configs_start_mode_earliest_to_assert.conf",
+ Collections.singletonList("topic=" + topic));
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+ }
+
+ @TestTemplate
+ public void testSourceKafkaTablesConfigsLatestStartMode(TestContainer
container)
+ throws IOException, InterruptedException {
+ String topic = startModeTopic(container, "latest");
+ createKafkaTopic(topic);
+ generateSimpleJsonRecords(topic, 0, 10);
+
+ Container.ExecResult execResult =
+ container.executeJob(
+
"/kafka/kafkasource_tables_configs_start_mode_latest_to_assert.conf",
+ Collections.singletonList("topic=" + topic));
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+ }
+
+ @TestTemplate
+ public void
testSourceKafkaTablesConfigsGroupOffsetsStartMode(TestContainer container)
+ throws IOException, InterruptedException {
+ String topic = startModeTopic(container, "group_offsets");
+ String consumerGroup = "SeaTunnel-Start-Mode-Tables-Group-Offsets";
+ createKafkaTopic(topic);
+ generateSimpleJsonRecords(topic, 0, 5);
+ commitOffset(topic, consumerGroup);
+ generateSimpleJsonRecords(topic, 5, 10);
+
+ Container.ExecResult execResult =
+ container.executeJob(
+
"/kafka/kafkasource_tables_configs_start_mode_group_offsets_to_assert.conf",
+ Collections.singletonList("topic=" + topic));
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+ }
+
+ @TestTemplate
+ public void testSourceKafkaTablesConfigsTimestampStartMode(TestContainer
container)
+ throws IOException, InterruptedException {
+ String topic = startModeTopic(container, "timestamp");
+ createKafkaTopic(topic);
+ generateSimpleJsonRecordsWithTimestamp(topic, 0, 4, 1738395839000L);
+ generateSimpleJsonRecordsWithTimestamp(topic, 4, 10, 1738395840000L);
+
+ Container.ExecResult execResult =
+ container.executeJob(
+
"/kafka/kafkasource_tables_configs_start_mode_timestamp_to_assert.conf",
+ Collections.singletonList("topic=" + topic));
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+ }
+
+ @TestTemplate
+ public void
testSourceKafkaTablesConfigsSpecificOffsetsStartMode(TestContainer container)
+ throws IOException, InterruptedException {
+ String topic = fixedStartModeTopic(container, "specific_offsets");
+ createKafkaTopic(topic);
+ generateSimpleJsonRecords(topic, 0, 10);
+
+ Container.ExecResult execResult =
+ container.executeJob(
+
"/kafka/kafkasource_tables_configs_start_mode_specific_offsets_to_assert.conf",
+ Collections.singletonList("topic=" + topic));
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+ }
+
+ private String fixedStartModeTopic(TestContainer container, String
startMode) {
+ return "test_topic_start_mode_"
+ + container.identifier().name().toLowerCase(Locale.ROOT)
+ + "_"
+ + startMode;
+ }
+
+ private String startModeTopic(TestContainer container, String startMode) {
+ return "test_topic_start_mode_"
+ + container.identifier().name().toLowerCase(Locale.ROOT)
+ + "_"
+ + startMode
+ + "_"
+ + startModeTestSequence.incrementAndGet();
+ }
+
@TestTemplate
public void testSourceKafkaStartConfig(TestContainer container)
throws IOException, InterruptedException {
@@ -1135,6 +1229,31 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
testKafkaGroupOffsetsToConsole(container);
}
+ private void generateSimpleJsonRecords(String topic, int startInclusive,
int endExclusive) {
+ generateSimpleJsonRecordsWithTimestamp(topic, startInclusive,
endExclusive, null);
+ }
+
+ private void generateSimpleJsonRecordsWithTimestamp(
+ String topic, int startInclusive, int endExclusive, Long
timestamp) {
+ try {
+ for (int i = startInclusive; i < endExclusive; i++) {
+ byte[] value = ("{\"id\":" + i +
"}").getBytes(StandardCharsets.UTF_8);
+ ProducerRecord<byte[], byte[]> record =
+ timestamp == null
+ ? new ProducerRecord<>(topic, null, value)
+ : new ProducerRecord<>(topic, null, timestamp,
null, value);
+ producer.send(record).get();
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while writing Kafka
records to " + topic, e);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to write Kafka start mode
records to " + topic, e);
+ } finally {
+ producer.flush();
+ }
+ }
+
public void commitOffset(String topic, String groupId) {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaContainer.getBootstrapServers());
@@ -1712,7 +1831,7 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
return null;
});
// wait for data written to kafka
- given().pollDelay(60, SECONDS)
+ given().pollDelay(120, SECONDS)
.pollInterval(5, SECONDS)
.await()
.atMost(5, MINUTES)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_tables_configs_start_mode_earliest_to_assert.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_tables_configs_start_mode_earliest_to_assert.conf
new file mode 100644
index 0000000000..cfe7b7692f
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_tables_configs_start_mode_earliest_to_assert.conf
@@ -0,0 +1,64 @@
+#
+# 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 = "BATCH"
+}
+
+source {
+ Kafka {
+ bootstrap.servers = "kafkaCluster:9092"
+ format = json
+ tables_configs = [
+ {
+ topic = "${topic}"
+ start_mode = earliest
+ schema = {
+ table = "kafka_start_mode.earliest"
+ fields {
+ id = bigint
+ }
+ }
+ }
+ ]
+ }
+}
+
+sink {
+ Assert {
+ rules = {
+ tables_configs = [
+ {
+ table_path = "kafka_start_mode.earliest"
+ row_rules = [
+ { rule_type = MIN_ROW, rule_value = 10 },
+ { rule_type = MAX_ROW, rule_value = 10 }
+ ]
+ field_rules = [{
+ field_name = id
+ field_type = bigint
+ field_value = [
+ { rule_type = MIN, rule_value = 0 },
+ { rule_type = MAX, rule_value = 9 }
+ ]
+ }]
+ }
+ ]
+ }
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_tables_configs_start_mode_group_offsets_to_assert.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_tables_configs_start_mode_group_offsets_to_assert.conf
new file mode 100644
index 0000000000..ab6e7b8cb2
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_tables_configs_start_mode_group_offsets_to_assert.conf
@@ -0,0 +1,65 @@
+#
+# 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 = "BATCH"
+}
+
+source {
+ Kafka {
+ bootstrap.servers = "kafkaCluster:9092"
+ format = json
+ consumer.group = "SeaTunnel-Start-Mode-Tables-Group-Offsets"
+ tables_configs = [
+ {
+ topic = "${topic}"
+ start_mode = group_offsets
+ schema = {
+ table = "kafka_start_mode.group_offsets"
+ fields {
+ id = bigint
+ }
+ }
+ }
+ ]
+ }
+}
+
+sink {
+ Assert {
+ rules = {
+ tables_configs = [
+ {
+ table_path = "kafka_start_mode.group_offsets"
+ row_rules = [
+ { rule_type = MIN_ROW, rule_value = 5 },
+ { rule_type = MAX_ROW, rule_value = 5 }
+ ]
+ field_rules = [{
+ field_name = id
+ field_type = bigint
+ field_value = [
+ { rule_type = MIN, rule_value = 5 },
+ { rule_type = MAX, rule_value = 9 }
+ ]
+ }]
+ }
+ ]
+ }
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_tables_configs_start_mode_latest_to_assert.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_tables_configs_start_mode_latest_to_assert.conf
new file mode 100644
index 0000000000..4d32b17800
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_tables_configs_start_mode_latest_to_assert.conf
@@ -0,0 +1,56 @@
+#
+# 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 = "BATCH"
+}
+
+source {
+ Kafka {
+ bootstrap.servers = "kafkaCluster:9092"
+ format = json
+ tables_configs = [
+ {
+ topic = "${topic}"
+ start_mode = latest
+ schema = {
+ table = "kafka_start_mode.latest"
+ fields {
+ id = bigint
+ }
+ }
+ }
+ ]
+ }
+}
+
+sink {
+ Assert {
+ rules = {
+ tables_configs = [
+ {
+ table_path = "kafka_start_mode.latest"
+ row_rules = [
+ { rule_type = MIN_ROW, rule_value = 0 },
+ { rule_type = MAX_ROW, rule_value = 0 }
+ ]
+ }
+ ]
+ }
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_tables_configs_start_mode_specific_offsets_to_assert.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_tables_configs_start_mode_specific_offsets_to_assert.conf
new file mode 100644
index 0000000000..d7290a5032
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_tables_configs_start_mode_specific_offsets_to_assert.conf
@@ -0,0 +1,76 @@
+#
+# 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 = "BATCH"
+}
+
+source {
+ Kafka {
+ bootstrap.servers = "kafkaCluster:9092"
+ format = json
+ tables_configs = [
+ {
+ topic = "${topic}"
+ start_mode = specific_offsets
+ start_mode.offsets = {
+ test_topic_start_mode_flink_1_13_specific_offsets-0 = 5
+ test_topic_start_mode_flink_1_14_specific_offsets-0 = 5
+ test_topic_start_mode_flink_1_15_specific_offsets-0 = 5
+ test_topic_start_mode_flink_1_16_specific_offsets-0 = 5
+ test_topic_start_mode_flink_1_17_specific_offsets-0 = 5
+ test_topic_start_mode_flink_1_18_specific_offsets-0 = 5
+ test_topic_start_mode_flink_1_20_specific_offsets-0 = 5
+ test_topic_start_mode_spark_2_4_specific_offsets-0 = 5
+ test_topic_start_mode_spark_3_3_specific_offsets-0 = 5
+ test_topic_start_mode_seatunnel_specific_offsets-0 = 5
+ }
+ schema = {
+ table = "kafka_start_mode.specific_offsets"
+ fields {
+ id = bigint
+ }
+ }
+ }
+ ]
+ }
+}
+
+sink {
+ Assert {
+ rules = {
+ tables_configs = [
+ {
+ table_path = "kafka_start_mode.specific_offsets"
+ row_rules = [
+ { rule_type = MIN_ROW, rule_value = 5 },
+ { rule_type = MAX_ROW, rule_value = 5 }
+ ]
+ field_rules = [{
+ field_name = id
+ field_type = bigint
+ field_value = [
+ { rule_type = MIN, rule_value = 5 },
+ { rule_type = MAX, rule_value = 9 }
+ ]
+ }]
+ }
+ ]
+ }
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_tables_configs_start_mode_timestamp_to_assert.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_tables_configs_start_mode_timestamp_to_assert.conf
new file mode 100644
index 0000000000..b4599877b6
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafkasource_tables_configs_start_mode_timestamp_to_assert.conf
@@ -0,0 +1,65 @@
+#
+# 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 = "BATCH"
+}
+
+source {
+ Kafka {
+ bootstrap.servers = "kafkaCluster:9092"
+ format = json
+ tables_configs = [
+ {
+ topic = "${topic}"
+ start_mode = timestamp
+ start_mode.timestamp = 1738395840000
+ schema = {
+ table = "kafka_start_mode.timestamp"
+ fields {
+ id = bigint
+ }
+ }
+ }
+ ]
+ }
+}
+
+sink {
+ Assert {
+ rules = {
+ tables_configs = [
+ {
+ table_path = "kafka_start_mode.timestamp"
+ row_rules = [
+ { rule_type = MIN_ROW, rule_value = 6 },
+ { rule_type = MAX_ROW, rule_value = 6 }
+ ]
+ field_rules = [{
+ field_name = id
+ field_type = bigint
+ field_value = [
+ { rule_type = MIN, rule_value = 4 },
+ { rule_type = MAX, rule_value = 9 }
+ ]
+ }]
+ }
+ ]
+ }
+ }
+}
