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-12352-2a8abf1e65971dd9c91b2ef69b2c90b747b2e0fb in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 6eea1da1a8c10e1ed048ce59db34d2f13ae685dc Author: Nikhil kumar <[email protected]> AuthorDate: Wed Sep 23 10:23:15 2026 +0000 [Improve][Connector-V2] Validate SLS required options (#12352) --- docs/en/connectors/sink/Sls.md | 10 +-- docs/en/connectors/source/Sls.md | 10 +-- docs/zh/connectors/sink/Sls.md | 10 +-- docs/zh/connectors/source/Sls.md | 10 +-- .../seatunnel/sls/sink/SlsSinkFactory.java | 13 +-- .../seatunnel/sls/source/SlsSourceFactory.java | 13 +-- .../connectors/seatunnel/sls/SlsFactoryTest.java | 94 +++++++++++++++++++++- 7 files changed, 127 insertions(+), 33 deletions(-) diff --git a/docs/en/connectors/sink/Sls.md b/docs/en/connectors/sink/Sls.md index 3c18322d45..77bc18be18 100644 --- a/docs/en/connectors/sink/Sls.md +++ b/docs/en/connectors/sink/Sls.md @@ -34,11 +34,11 @@ Maven central repository. | Name | Type | Required | Default | Description | |-------------------|--------|----------|--------------------|---------------------------------------------------------------------------------------------------------| -| endpoint | String | Yes | - | Alibaba Cloud SLS endpoint, for example `cn-hangzhou.log.aliyuncs.com` or an intranet endpoint. | -| project | String | Yes | - | [Alibaba Cloud SLS project](https://help.aliyun.com/zh/sls/user-guide/manage-a-project). | -| logstore | String | Yes | - | [Alibaba Cloud SLS logstore](https://help.aliyun.com/zh/sls/user-guide/manage-a-logstore). | -| access_key_id | String | Yes | - | Alibaba Cloud AccessKey ID. | -| access_key_secret | String | Yes | - | Alibaba Cloud AccessKey secret. | +| endpoint | String | Yes | - | Alibaba Cloud SLS endpoint, for example `cn-hangzhou.log.aliyuncs.com` or an intranet endpoint. Must not be empty or whitespace-only. | +| project | String | Yes | - | [Alibaba Cloud SLS project](https://help.aliyun.com/zh/sls/user-guide/manage-a-project). Must not be empty or whitespace-only. | +| logstore | String | Yes | - | [Alibaba Cloud SLS logstore](https://help.aliyun.com/zh/sls/user-guide/manage-a-logstore). Must not be empty or whitespace-only. | +| access_key_id | String | Yes | - | Alibaba Cloud AccessKey ID. Must not be empty or whitespace-only. | +| access_key_secret | String | Yes | - | Alibaba Cloud AccessKey secret. Must not be empty or whitespace-only. | | source | String | No | `SeaTunnel-Source` | Source tag written to SLS log groups. | | topic | String | No | `SeaTunnel-Topic` | Topic tag written to SLS log groups. | diff --git a/docs/en/connectors/source/Sls.md b/docs/en/connectors/source/Sls.md index 161b00ca6a..0e03a529c0 100644 --- a/docs/en/connectors/source/Sls.md +++ b/docs/en/connectors/source/Sls.md @@ -43,11 +43,11 @@ Maven central repository. | Name | Type | Required | Default | Description | |-------------------------------------|-------------------------------------------|----------|--------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------| -| endpoint | String | Yes | - | Alibaba Cloud SLS endpoint, for example `cn-hangzhou.log.aliyuncs.com` or an intranet endpoint. | -| project | String | Yes | - | [Alibaba Cloud SLS project](https://help.aliyun.com/zh/sls/user-guide/manage-a-project). | -| logstore | String | Yes | - | [Alibaba Cloud SLS logstore](https://help.aliyun.com/zh/sls/user-guide/manage-a-logstore). | -| access_key_id | String | Yes | - | Alibaba Cloud AccessKey ID. | -| access_key_secret | String | Yes | - | Alibaba Cloud AccessKey secret. | +| endpoint | String | Yes | - | Alibaba Cloud SLS endpoint, for example `cn-hangzhou.log.aliyuncs.com` or an intranet endpoint. Must not be empty or whitespace-only. | +| project | String | Yes | - | [Alibaba Cloud SLS project](https://help.aliyun.com/zh/sls/user-guide/manage-a-project). Must not be empty or whitespace-only. | +| logstore | String | Yes | - | [Alibaba Cloud SLS logstore](https://help.aliyun.com/zh/sls/user-guide/manage-a-logstore). Must not be empty or whitespace-only. | +| access_key_id | String | Yes | - | Alibaba Cloud AccessKey ID. Must not be empty or whitespace-only. | +| access_key_secret | String | Yes | - | Alibaba Cloud AccessKey secret. Must not be empty or whitespace-only. | | start_mode | `earliest`, `group_cursor`, `latest` | No | `group_cursor` | Initial cursor mode. `earliest` starts from the beginning, `latest` starts from the end, and `group_cursor` uses the consumer group's checkpoint. | | consumer_group | String | No | `SeaTunnel-Consumer-Group` | SLS consumer group name. Use different values when separate jobs must keep independent cursors. | | auto_cursor_reset | `begin`, `end` | No | `end` | Cursor position used when `start_mode = group_cursor` but the consumer group has no checkpoint yet. | diff --git a/docs/zh/connectors/sink/Sls.md b/docs/zh/connectors/sink/Sls.md index e398948446..3985bf21d0 100644 --- a/docs/zh/connectors/sink/Sls.md +++ b/docs/zh/connectors/sink/Sls.md @@ -33,11 +33,11 @@ JSON，然后作为 SLS 日志项写入，日志内容的 key 为 `content`。 | 名称 | 类型 | 是否必填 | 默认值 | 描述 | |-------------------|--------|----------|--------------------|--------------------------------------------------------------------------------------------------------| -| endpoint | String | 是 | - | 阿里云 SLS 访问地址，例如 `cn-hangzhou.log.aliyuncs.com` 或内网访问地址。 | -| project | String | 是 | - | [阿里云 SLS Project](https://help.aliyun.com/zh/sls/user-guide/manage-a-project)。 | -| logstore | String | 是 | - | [阿里云 SLS Logstore](https://help.aliyun.com/zh/sls/user-guide/manage-a-logstore)。 | -| access_key_id | String | 是 | - | 阿里云 AccessKey ID。 | -| access_key_secret | String | 是 | - | 阿里云 AccessKey Secret。 | +| endpoint | String | 是 | - | 阿里云 SLS 访问地址，例如 `cn-hangzhou.log.aliyuncs.com` 或内网访问地址。值不能为空或仅包含空白字符。 | +| project | String | 是 | - | [阿里云 SLS Project](https://help.aliyun.com/zh/sls/user-guide/manage-a-project)。值不能为空或仅包含空白字符。 | +| logstore | String | 是 | - | [阿里云 SLS Logstore](https://help.aliyun.com/zh/sls/user-guide/manage-a-logstore)。值不能为空或仅包含空白字符。 | +| access_key_id | String | 是 | - | 阿里云 AccessKey ID。值不能为空或仅包含空白字符。 | +| access_key_secret | String | 是 | - | 阿里云 AccessKey Secret。值不能为空或仅包含空白字符。 | | source | String | 否 | `SeaTunnel-Source` | 写入 SLS log group 的 source 标记。 | | topic | String | 否 | `SeaTunnel-Topic` | 写入 SLS log group 的 topic 标记。 | diff --git a/docs/zh/connectors/source/Sls.md b/docs/zh/connectors/source/Sls.md index 54db0f1538..16424c3fa2 100644 --- a/docs/zh/connectors/source/Sls.md +++ b/docs/zh/connectors/source/Sls.md @@ -42,11 +42,11 @@ SLS shard 并行读取。流处理模式下，SeaTunnel 会在 checkpoint 完成 | 名称 | 类型 | 是否必填 | 默认值 | 描述 | |-------------------------------------|-------------------------------------------|----------|----------------------------|----------------------------------------------------------------------------------------------------------------| -| endpoint | String | 是 | - | 阿里云 SLS 访问地址，例如 `cn-hangzhou.log.aliyuncs.com` 或内网访问地址。 | -| project | String | 是 | - | [阿里云 SLS Project](https://help.aliyun.com/zh/sls/user-guide/manage-a-project)。 | -| logstore | String | 是 | - | [阿里云 SLS Logstore](https://help.aliyun.com/zh/sls/user-guide/manage-a-logstore)。 | -| access_key_id | String | 是 | - | 阿里云 AccessKey ID。 | -| access_key_secret | String | 是 | - | 阿里云 AccessKey Secret。 | +| endpoint | String | 是 | - | 阿里云 SLS 访问地址，例如 `cn-hangzhou.log.aliyuncs.com` 或内网访问地址。值不能为空或仅包含空白字符。 | +| project | String | 是 | - | [阿里云 SLS Project](https://help.aliyun.com/zh/sls/user-guide/manage-a-project)。值不能为空或仅包含空白字符。 | +| logstore | String | 是 | - | [阿里云 SLS Logstore](https://help.aliyun.com/zh/sls/user-guide/manage-a-logstore)。值不能为空或仅包含空白字符。| +| access_key_id | String | 是 | - | 阿里云 AccessKey ID。值不能为空或仅包含空白字符。 | +| access_key_secret | String | 是 | - | 阿里云 AccessKey Secret。值不能为空或仅包含空白字符。 | | start_mode | `earliest`, `group_cursor`, `latest` | 否 | `group_cursor` | 初始读取位置。`earliest` 从最早位置读取，`latest` 从最新位置读取，`group_cursor` 使用消费者组已提交的游标。 | | consumer_group | String | 否 | `SeaTunnel-Consumer-Group` | SLS 消费者组名称。不同任务如果需要独立保存读取位置，应该使用不同的消费者组。 | | auto_cursor_reset | `begin`, `end` | 否 | `end` | 当 `start_mode = group_cursor` 但消费者组还没有已提交游标时，使用的初始化位置。 | diff --git a/seatunnel-connectors-v2/connector-sls/src/main/java/org/apache/seatunnel/connectors/seatunnel/sls/sink/SlsSinkFactory.java b/seatunnel-connectors-v2/connector-sls/src/main/java/org/apache/seatunnel/connectors/seatunnel/sls/sink/SlsSinkFactory.java index 099da96baa..a869bc8a90 100644 --- a/seatunnel-connectors-v2/connector-sls/src/main/java/org/apache/seatunnel/connectors/seatunnel/sls/sink/SlsSinkFactory.java +++ b/seatunnel-connectors-v2/connector-sls/src/main/java/org/apache/seatunnel/connectors/seatunnel/sls/sink/SlsSinkFactory.java @@ -26,6 +26,8 @@ import org.apache.seatunnel.connectors.seatunnel.sls.config.SlsSinkOptions; import com.google.auto.service.AutoService; +import static org.apache.seatunnel.api.configuration.util.Conditions.notBlank; + @AutoService(Factory.class) public class SlsSinkFactory implements TableSinkFactory { @Override @@ -36,12 +38,13 @@ public class SlsSinkFactory implements TableSinkFactory { @Override public OptionRule optionRule() { return OptionRule.builder() + .required(SlsSinkOptions.ENDPOINT, notBlank(SlsSinkOptions.ENDPOINT)) + .required(SlsSinkOptions.PROJECT, notBlank(SlsSinkOptions.PROJECT)) + .required(SlsSinkOptions.LOGSTORE, notBlank(SlsSinkOptions.LOGSTORE)) + .required(SlsSinkOptions.ACCESS_KEY_ID, notBlank(SlsSinkOptions.ACCESS_KEY_ID)) .required( - SlsSinkOptions.ENDPOINT, - SlsSinkOptions.PROJECT, - SlsSinkOptions.LOGSTORE, - SlsSinkOptions.ACCESS_KEY_ID, - SlsSinkOptions.ACCESS_KEY_SECRET) + SlsSinkOptions.ACCESS_KEY_SECRET, + notBlank(SlsSinkOptions.ACCESS_KEY_SECRET)) .optional(SlsSinkOptions.SOURCE, SlsSinkOptions.TOPIC) .build(); } diff --git a/seatunnel-connectors-v2/connector-sls/src/main/java/org/apache/seatunnel/connectors/seatunnel/sls/source/SlsSourceFactory.java b/seatunnel-connectors-v2/connector-sls/src/main/java/org/apache/seatunnel/connectors/seatunnel/sls/source/SlsSourceFactory.java index 5e698bc498..bce62586d2 100644 --- a/seatunnel-connectors-v2/connector-sls/src/main/java/org/apache/seatunnel/connectors/seatunnel/sls/source/SlsSourceFactory.java +++ b/seatunnel-connectors-v2/connector-sls/src/main/java/org/apache/seatunnel/connectors/seatunnel/sls/source/SlsSourceFactory.java @@ -30,6 +30,8 @@ import com.google.auto.service.AutoService; import java.io.Serializable; +import static org.apache.seatunnel.api.configuration.util.Conditions.notBlank; + @AutoService(Factory.class) public class SlsSourceFactory implements TableSourceFactory { @Override @@ -45,12 +47,13 @@ public class SlsSourceFactory implements TableSourceFactory { @Override public OptionRule optionRule() { return OptionRule.builder() + .required(SlsSourceOptions.ENDPOINT, notBlank(SlsSourceOptions.ENDPOINT)) + .required(SlsSourceOptions.PROJECT, notBlank(SlsSourceOptions.PROJECT)) + .required(SlsSourceOptions.LOGSTORE, notBlank(SlsSourceOptions.LOGSTORE)) + .required(SlsSourceOptions.ACCESS_KEY_ID, notBlank(SlsSourceOptions.ACCESS_KEY_ID)) .required( - SlsSourceOptions.ENDPOINT, - SlsSourceOptions.PROJECT, - SlsSourceOptions.LOGSTORE, - SlsSourceOptions.ACCESS_KEY_ID, - SlsSourceOptions.ACCESS_KEY_SECRET) + SlsSourceOptions.ACCESS_KEY_SECRET, + notBlank(SlsSourceOptions.ACCESS_KEY_SECRET)) .optional( SlsSourceOptions.BATCH_SIZE, SlsSourceOptions.START_MODE, diff --git a/seatunnel-connectors-v2/connector-sls/src/test/java/org/apache/seatunnel/connectors/seatunnel/sls/SlsFactoryTest.java b/seatunnel-connectors-v2/connector-sls/src/test/java/org/apache/seatunnel/connectors/seatunnel/sls/SlsFactoryTest.java index 84780c77dd..e8130e3c0c 100644 --- a/seatunnel-connectors-v2/connector-sls/src/test/java/org/apache/seatunnel/connectors/seatunnel/sls/SlsFactoryTest.java +++ b/seatunnel-connectors-v2/connector-sls/src/test/java/org/apache/seatunnel/connectors/seatunnel/sls/SlsFactoryTest.java @@ -17,17 +17,105 @@ package org.apache.seatunnel.connectors.seatunnel.sls; +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.sls.config.SlsBaseOptions; import org.apache.seatunnel.connectors.seatunnel.sls.sink.SlsSinkFactory; import org.apache.seatunnel.connectors.seatunnel.sls.source.SlsSourceFactory; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + public class SlsFactoryTest { + private static final OptionRule SOURCE_OPTION_RULE = new SlsSourceFactory().optionRule(); + private static final OptionRule SINK_OPTION_RULE = new SlsSinkFactory().optionRule(); + private static final List<String> REQUIRED_STRING_KEYS = + Arrays.asList( + SlsBaseOptions.ENDPOINT.key(), + SlsBaseOptions.PROJECT.key(), + SlsBaseOptions.LOGSTORE.key(), + SlsBaseOptions.ACCESS_KEY_ID.key(), + SlsBaseOptions.ACCESS_KEY_SECRET.key()); + + @Test + void testValidSourceConfiguration() { + assertValidRequiredStrings(SOURCE_OPTION_RULE, "SlsSource"); + } + + @Test + void testInvalidSourceConfiguration() { + assertInvalidRequiredStrings(SOURCE_OPTION_RULE, "SlsSource"); + } + + @Test + void testValidSinkConfiguration() { + assertValidRequiredStrings(SINK_OPTION_RULE, "SlsSink"); + } + @Test - void optionRule() { - Assertions.assertNotNull((new SlsSourceFactory()).optionRule()); - Assertions.assertNotNull((new SlsSinkFactory()).optionRule()); + void testInvalidSinkConfiguration() { + assertInvalidRequiredStrings(SINK_OPTION_RULE, "SlsSink"); + } + + private void assertValidRequiredStrings(OptionRule optionRule, String factoryName) { + Assertions.assertDoesNotThrow(() -> validate(validConfig(), optionRule, factoryName)); + + Map<String, Object> valuesWithSurroundingSpaces = validConfig(); + for (String key : REQUIRED_STRING_KEYS) { + valuesWithSurroundingSpaces.put(key, " " + valuesWithSurroundingSpaces.get(key) + " "); + } + Assertions.assertDoesNotThrow( + () -> validate(valuesWithSurroundingSpaces, optionRule, factoryName)); + } + + private void assertInvalidRequiredStrings(OptionRule optionRule, String factoryName) { + for (String key : REQUIRED_STRING_KEYS) { + Map<String, Object> missing = validConfig(); + missing.remove(key); + assertInvalid(missing, optionRule, factoryName, key); + + Map<String, Object> empty = validConfig(); + empty.put(key, ""); + assertInvalid(empty, optionRule, factoryName, key); + + Map<String, Object> whitespaceOnly = validConfig(); + whitespaceOnly.put(key, " \t\n "); + assertInvalid(whitespaceOnly, optionRule, factoryName, key); + } + } + + private void assertInvalid( + Map<String, Object> config, + OptionRule optionRule, + String factoryName, + String optionKey) { + Assertions.assertThrows( + OptionValidationException.class, + () -> validate(config, optionRule, factoryName), + optionKey); + } + + private void validate(Map<String, Object> config, OptionRule optionRule, String factoryName) { + ReadonlyConfig readonlyConfig = ReadonlyConfig.fromMap(config); + ConfigValidator.validateUnknownKeys(readonlyConfig, optionRule, factoryName); + ConfigValidator.of(readonlyConfig).validate(optionRule); + } + + private Map<String, Object> validConfig() { + Map<String, Object> config = new HashMap<>(); + config.put(SlsBaseOptions.ENDPOINT.key(), "cn-hangzhou.log.aliyuncs.com"); + config.put(SlsBaseOptions.PROJECT.key(), "seatunnel-project"); + config.put(SlsBaseOptions.LOGSTORE.key(), "seatunnel-logstore"); + config.put(SlsBaseOptions.ACCESS_KEY_ID.key(), "access-key-id"); + config.put(SlsBaseOptions.ACCESS_KEY_SECRET.key(), "access-key-secret"); + return config; } }
