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 ea5f089674 [Improve][Connector-V2][Easysearch] Declarative optionRule
value constraints for source and sink (#12414)
ea5f089674 is described below
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));
}
}
