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-12414-4cf17541df318472b5fffdfcbaab2b1a33ec31c8 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit ea5f089674ea782c9399503258a0940459cf7abb Author: TianHengZhuang <[email protected]> AuthorDate: Tue Sep 29 13:35:17 2026 +0000 [Improve][Connector-V2][Easysearch] Declarative optionRule value constraints for source and sink (#12414) --- .../easysearch/sink/EasysearchSinkFactory.java | 16 ++- .../easysearch/source/EasysearchSourceFactory.java | 12 +- .../easysearch/EasysearchFactoryTest.java | 122 ++++++++++++++++++++- 3 files changed, 143 insertions(+), 7 deletions(-) diff --git a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/sink/EasysearchSinkFactory.java b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/sink/EasysearchSinkFactory.java index 5ab636605e..ec38d23166 100644 --- a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/sink/EasysearchSinkFactory.java +++ b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/sink/EasysearchSinkFactory.java @@ -17,6 +17,7 @@ package org.apache.seatunnel.connectors.seatunnel.easysearch.sink; +import org.apache.seatunnel.api.configuration.util.Conditions; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.table.connector.TableSink; import org.apache.seatunnel.api.table.factory.Factory; @@ -36,14 +37,23 @@ public class EasysearchSinkFactory implements TableSinkFactory { @Override public OptionRule optionRule() { return OptionRule.builder() - .required(EasysearchSinkOptions.HOSTS, EasysearchSinkOptions.INDEX) + .required( + EasysearchSinkOptions.HOSTS, + Conditions.notEmpty(EasysearchSinkOptions.HOSTS)) + .required( + EasysearchSinkOptions.INDEX, + Conditions.notBlank(EasysearchSinkOptions.INDEX)) + .optional( + EasysearchSinkOptions.MAX_BATCH_SIZE, + Conditions.greaterThan(EasysearchSinkOptions.MAX_BATCH_SIZE, 0)) + .optional( + EasysearchSinkOptions.MAX_RETRY_COUNT, + Conditions.greaterOrEqual(EasysearchSinkOptions.MAX_RETRY_COUNT, 0)) .optional( EasysearchSinkOptions.USERNAME, EasysearchSinkOptions.PASSWORD, EasysearchSinkOptions.PRIMARY_KEYS, EasysearchSinkOptions.KEY_DELIMITER, - EasysearchSinkOptions.MAX_RETRY_COUNT, - EasysearchSinkOptions.MAX_BATCH_SIZE, EasysearchSinkOptions.TLS_VERIFY_CERTIFICATE, EasysearchSinkOptions.TLS_VERIFY_HOSTNAME, EasysearchSinkOptions.TLS_KEY_STORE_PATH, diff --git a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/source/EasysearchSourceFactory.java b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/source/EasysearchSourceFactory.java index e766db2640..38a0cb9326 100644 --- a/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/source/EasysearchSourceFactory.java +++ b/seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/source/EasysearchSourceFactory.java @@ -20,6 +20,7 @@ package org.apache.seatunnel.connectors.seatunnel.easysearch.source; import org.apache.seatunnel.shade.com.google.common.collect.Lists; 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; @@ -59,12 +60,19 @@ public class EasysearchSourceFactory implements TableSourceFactory { @Override public OptionRule optionRule() { return OptionRule.builder() - .required(EasysearchSourceOptions.HOSTS, EasysearchSourceOptions.INDEX) + .required( + EasysearchSourceOptions.HOSTS, + Conditions.notEmpty(EasysearchSourceOptions.HOSTS)) + .required( + EasysearchSourceOptions.INDEX, + Conditions.notBlank(EasysearchSourceOptions.INDEX)) + .optional( + EasysearchSourceOptions.SCROLL_SIZE, + Conditions.greaterThan(EasysearchSourceOptions.SCROLL_SIZE, 0)) .optional( EasysearchSourceOptions.USERNAME, EasysearchSourceOptions.PASSWORD, EasysearchSourceOptions.SCROLL_TIME, - EasysearchSourceOptions.SCROLL_SIZE, EasysearchSourceOptions.QUERY, EasysearchSourceOptions.TLS_VERIFY_CERTIFICATE, EasysearchSourceOptions.TLS_VERIFY_HOSTNAME, diff --git a/seatunnel-connectors-v2/connector-easysearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/easysearch/EasysearchFactoryTest.java b/seatunnel-connectors-v2/connector-easysearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/easysearch/EasysearchFactoryTest.java index 6da684a5d0..c0920510ec 100644 --- a/seatunnel-connectors-v2/connector-easysearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/easysearch/EasysearchFactoryTest.java +++ b/seatunnel-connectors-v2/connector-easysearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/easysearch/EasysearchFactoryTest.java @@ -17,17 +17,135 @@ package org.apache.seatunnel.connectors.seatunnel.easysearch; +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.configuration.util.ConfigValidator; +import org.apache.seatunnel.api.configuration.util.OptionValidationException; import org.apache.seatunnel.connectors.seatunnel.easysearch.sink.EasysearchSinkFactory; import org.apache.seatunnel.connectors.seatunnel.easysearch.source.EasysearchSourceFactory; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + public class EasysearchFactoryTest { + private final EasysearchSourceFactory sourceFactory = new EasysearchSourceFactory(); + private final EasysearchSinkFactory sinkFactory = new EasysearchSinkFactory(); + + private static Map<String, Object> validSourceConfig() { + Map<String, Object> map = new HashMap<>(); + map.put("hosts", Collections.singletonList("http://127.0.0.1:9200";)); + map.put("index", "demo"); + map.put("source", Collections.singletonList("id")); + return map; + } + + private static Map<String, Object> validSinkConfig() { + Map<String, Object> map = new HashMap<>(); + map.put("hosts", Collections.singletonList("http://127.0.0.1:9200";)); + map.put("index", "demo"); + return map; + } + + private void validateSource(Map<String, Object> map) { + ConfigValidator.of(ReadonlyConfig.fromMap(map)).validate(sourceFactory.optionRule()); + } + + private void validateSink(Map<String, Object> map) { + ConfigValidator.of(ReadonlyConfig.fromMap(map)).validate(sinkFactory.optionRule()); + } + @Test void optionRule() { - Assertions.assertNotNull((new EasysearchSourceFactory()).optionRule()); - Assertions.assertNotNull((new EasysearchSinkFactory()).optionRule()); + Assertions.assertNotNull(sourceFactory.optionRule()); + Assertions.assertNotNull(sinkFactory.optionRule()); + } + + @Test + void validSourceConfigPassesValidation() { + validateSource(validSourceConfig()); + } + + @Test + void validSinkConfigPassesValidation() { + validateSink(validSinkConfig()); + } + + @Test + void validNumericBoundsPassValidation() { + Map<String, Object> source = validSourceConfig(); + source.put("scroll_size", 1); + validateSource(source); + + Map<String, Object> sink = validSinkConfig(); + sink.put("max_batch_size", 1); + sink.put("max_retry_count", 0); + validateSink(sink); + } + + @Test + void blankIndexIsRejectedForSourceAndSink() { + Map<String, Object> source = validSourceConfig(); + source.put("index", " "); + Assertions.assertThrows(OptionValidationException.class, () -> validateSource(source)); + + Map<String, Object> sink = validSinkConfig(); + sink.put("index", ""); + Assertions.assertThrows(OptionValidationException.class, () -> validateSink(sink)); + } + + @Test + void emptyHostsIsRejectedForSourceAndSink() { + Map<String, Object> source = validSourceConfig(); + source.put("hosts", Collections.emptyList()); + Assertions.assertThrows(OptionValidationException.class, () -> validateSource(source)); + + Map<String, Object> sink = validSinkConfig(); + sink.put("hosts", Collections.emptyList()); + Assertions.assertThrows(OptionValidationException.class, () -> validateSink(sink)); + } + + @Test + void missingHostsOrIndexIsRejected() { + Map<String, Object> source = validSourceConfig(); + source.remove("hosts"); + Assertions.assertThrows(OptionValidationException.class, () -> validateSource(source)); + + Map<String, Object> sink = validSinkConfig(); + sink.remove("index"); + Assertions.assertThrows(OptionValidationException.class, () -> validateSink(sink)); + } + + @Test + void nonPositiveScrollSizeIsRejected() { + Map<String, Object> source = validSourceConfig(); + source.put("scroll_size", 0); + Assertions.assertThrows(OptionValidationException.class, () -> validateSource(source)); + } + + @Test + void nonPositiveMaxBatchSizeIsRejected() { + Map<String, Object> sink = validSinkConfig(); + sink.put("max_batch_size", 0); + Assertions.assertThrows(OptionValidationException.class, () -> validateSink(sink)); + } + + @Test + void negativeMaxRetryCountIsRejected() { + Map<String, Object> sink = validSinkConfig(); + sink.put("max_retry_count", -1); + Assertions.assertThrows(OptionValidationException.class, () -> validateSink(sink)); + } + + @Test + void exclusiveSourceAndSchemaRejected() { + Map<String, Object> map = validSourceConfig(); + Map<String, Object> schema = new HashMap<>(); + schema.put("fields", new HashMap<String, Object>()); + map.put("schema", schema); + Assertions.assertThrows(OptionValidationException.class, () -> validateSource(map)); } }
