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-12315-3f04cb6176ab59bb9b5ec0cc2dc0b3b4a2de9a78 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 6fac4cc44182b4a998fc351c639c4b120592baf7 Author: Nikhil kumar <[email protected]> AuthorDate: Thu Sep 24 01:59:46 2026 +0000 [Improve][Connector-V2] Migrate SensorsData validation to OptionRule (#12315) --- .../sdk/sink/SensorsDataSDKSinkFactory.java | 6 ++ .../sensorsdata/sdk/SensorsDataSDKFactoryTest.java | 71 +++++++++++++++++++++- 2 files changed, 76 insertions(+), 1 deletion(-) diff --git a/seatunnel-connectors-v2/connector-sensorsdata/src/main/java/org/apache/seatunnel/connectors/sensorsdata/sdk/sink/SensorsDataSDKSinkFactory.java b/seatunnel-connectors-v2/connector-sensorsdata/src/main/java/org/apache/seatunnel/connectors/sensorsdata/sdk/sink/SensorsDataSDKSinkFactory.java index d965a4e400..0d002205d5 100644 --- a/seatunnel-connectors-v2/connector-sensorsdata/src/main/java/org/apache/seatunnel/connectors/sensorsdata/sdk/sink/SensorsDataSDKSinkFactory.java +++ b/seatunnel-connectors-v2/connector-sensorsdata/src/main/java/org/apache/seatunnel/connectors/sensorsdata/sdk/sink/SensorsDataSDKSinkFactory.java @@ -31,6 +31,8 @@ import org.apache.seatunnel.connectors.sensorsdata.sdk.config.SensorsDataSDKSink import com.google.auto.service.AutoService; +import static org.apache.seatunnel.api.configuration.util.Conditions.notBlank; + @AutoService(Factory.class) public class SensorsDataSDKSinkFactory implements TableSinkFactory { @Override @@ -41,9 +43,13 @@ public class SensorsDataSDKSinkFactory implements TableSinkFactory { @Override public OptionRule optionRule() { return SensorsDataBaseOptionRules.getBaseOptionRuleBuilder() + .required( + SensorsDataSDKSinkOptions.SERVER_URL, + notBlank(SensorsDataSDKSinkOptions.SERVER_URL)) .optional( SensorsDataSDKSinkOptions.BULK_SIZE, SensorsDataSDKSinkOptions.MAX_CACHE_ROW_SIZE, + SensorsDataSDKSinkOptions.CONSUMER, SensorsDataOptions.SKIP_ERROR_RECORD, SensorsDataSDKSinkOptions.INSTANT_EVENT_LIST) .build(); diff --git a/seatunnel-connectors-v2/connector-sensorsdata/src/test/java/org/apache/seatunnel/connectors/sensorsdata/sdk/SensorsDataSDKFactoryTest.java b/seatunnel-connectors-v2/connector-sensorsdata/src/test/java/org/apache/seatunnel/connectors/sensorsdata/sdk/SensorsDataSDKFactoryTest.java index 3537e61773..71f9239638 100644 --- a/seatunnel-connectors-v2/connector-sensorsdata/src/test/java/org/apache/seatunnel/connectors/sensorsdata/sdk/SensorsDataSDKFactoryTest.java +++ b/seatunnel-connectors-v2/connector-sensorsdata/src/test/java/org/apache/seatunnel/connectors/sensorsdata/sdk/SensorsDataSDKFactoryTest.java @@ -17,15 +17,84 @@ package org.apache.seatunnel.connectors.sensorsdata.sdk; +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.sensorsdata.format.config.SensorsDataOptions; +import org.apache.seatunnel.connectors.sensorsdata.sdk.config.SensorsDataSDKSinkOptions; import org.apache.seatunnel.connectors.sensorsdata.sdk.sink.SensorsDataSDKSinkFactory; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.HashMap; +import java.util.Map; + class SensorsDataSDKFactoryTest { + private static final OptionRule OPTION_RULE = new SensorsDataSDKSinkFactory().optionRule(); + @Test void optionRule() { - Assertions.assertNotNull((new SensorsDataSDKSinkFactory()).optionRule()); + Assertions.assertNotNull(OPTION_RULE); + } + + @Test + void testValidServerUrlAndConsumerConfigurations() { + Map<String, Object> batchConfig = validConfig(); + batchConfig.put(SensorsDataSDKSinkOptions.CONSUMER.key(), "batch"); + Assertions.assertDoesNotThrow(() -> validate(batchConfig)); + + Map<String, Object> consoleConfig = validConfig(); + consoleConfig.put(SensorsDataSDKSinkOptions.CONSUMER.key(), "console"); + Assertions.assertDoesNotThrow(() -> validate(consoleConfig)); + } + + @Test + void testMissingServerUrlRejected() { + Map<String, Object> config = validConfig(); + config.remove(SensorsDataSDKSinkOptions.SERVER_URL.key()); + + Assertions.assertThrows(OptionValidationException.class, () -> validate(config)); + } + + @Test + void testEmptyServerUrlRejected() { + Map<String, Object> config = validConfig(); + config.put(SensorsDataSDKSinkOptions.SERVER_URL.key(), ""); + + Assertions.assertThrows(OptionValidationException.class, () -> validate(config)); + } + + @Test + void testWhitespaceOnlyServerUrlRejected() { + Map<String, Object> config = validConfig(); + config.put(SensorsDataSDKSinkOptions.SERVER_URL.key(), " \t"); + + Assertions.assertThrows(OptionValidationException.class, () -> validate(config)); + } + + @Test + void testOmittedConsumerDefaultsToBatch() { + ReadonlyConfig config = validate(validConfig()); + + Assertions.assertEquals("batch", config.get(SensorsDataSDKSinkOptions.CONSUMER)); + } + + private ReadonlyConfig validate(Map<String, Object> config) { + ReadonlyConfig readonlyConfig = ReadonlyConfig.fromMap(config); + ConfigValidator.of(readonlyConfig).validate(OPTION_RULE); + return readonlyConfig; + } + + private Map<String, Object> validConfig() { + Map<String, Object> config = new HashMap<>(); + config.put( + SensorsDataSDKSinkOptions.SERVER_URL.key(), + "https://localhost:8106/sa?project=default";); + config.put(SensorsDataOptions.ENTITY_NAME.key(), "items"); + config.put(SensorsDataOptions.RECORD_TYPE.key(), "users"); + return config; } }
