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-11881-d57a0f685860363468afda9d4db89236e2c354b9 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 5acb54618d05263c59e2aacd66597fe015a53213 Author: Rohan Gaikwad <[email protected]> AuthorDate: Sun Aug 30 16:24:42 2026 +0000 [Improve][Connector-V2][File] Migrate HDFS and S3 sink validation to declarative OptionRule (#11881) --- .../file/hdfs/sink/HdfsFileSinkFactory.java | 15 ------- .../seatunnel/file/hdfs/HdfsFileFactoryTest.java | 26 ++++++++++++ .../seatunnel/file/s3/sink/S3FileSink.java | 21 ---------- .../seatunnel/file/s3/S3FileFactoryTest.java | 47 ++++++++++++++++++++++ 4 files changed, 73 insertions(+), 36 deletions(-) diff --git a/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/sink/HdfsFileSinkFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/sink/HdfsFileSinkFactory.java index 336d4cfaa8..46f3f40b69 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/sink/HdfsFileSinkFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/sink/HdfsFileSinkFactory.java @@ -19,7 +19,6 @@ package org.apache.seatunnel.connectors.seatunnel.file.hdfs.sink; import org.apache.seatunnel.shade.com.typesafe.config.Config; -import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode; import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.options.SinkConnectorCommonOptions; @@ -28,13 +27,9 @@ import org.apache.seatunnel.api.table.connector.TableSink; import org.apache.seatunnel.api.table.factory.Factory; import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext; import org.apache.seatunnel.api.table.type.SeaTunnelRow; -import org.apache.seatunnel.common.config.CheckConfigUtil; -import org.apache.seatunnel.common.config.CheckResult; -import org.apache.seatunnel.common.constants.PluginType; import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat; import org.apache.seatunnel.connectors.seatunnel.file.config.FileSystemType; import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf; -import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException; import org.apache.seatunnel.connectors.seatunnel.file.factory.BaseMultipleTableFileSinkFactory; import org.apache.seatunnel.connectors.seatunnel.file.hdfs.config.HdfsFileSinkOptions; import org.apache.seatunnel.connectors.seatunnel.file.sink.commit.FileAggregatedCommitInfo; @@ -144,16 +139,6 @@ public class HdfsFileSinkFactory extends BaseMultipleTableFileSinkFactory { public HadoopConf initHadoopConf(ReadonlyConfig readonlyConfig) { Config pluginConfig = readonlyConfig.toConfig(); - CheckResult result = - CheckConfigUtil.checkAllExists(readonlyConfig.toConfig(), FS_DEFAULT_NAME_KEY); - if (!result.isSuccess()) { - throw new FileConnectorException( - SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED, - String.format( - "PluginName: %s, PluginType: %s, Message: %s", - factoryIdentifier(), PluginType.SINK, result.getMsg())); - } - HadoopConf hadoopConf = new HadoopConf(pluginConfig.getString(FS_DEFAULT_NAME_KEY)); if (pluginConfig.hasPath(HdfsFileSinkOptions.HDFS_SITE_PATH.key())) { diff --git a/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/HdfsFileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/HdfsFileFactoryTest.java index 1d9ed3c988..f52d8abe22 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/HdfsFileFactoryTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/HdfsFileFactoryTest.java @@ -99,6 +99,32 @@ class HdfsFileFactoryTest { Assertions.assertDoesNotThrow(() -> validate(backupConfig, optionRule)); } + @Test + void sinkOptionRuleRequiresDefaultFs() { + OptionRule optionRule = (new HdfsFileSinkFactory()).optionRule(); + Map<String, Object> config = new HashMap<>(); + config.put(FileBaseOptions.FILE_PATH.key(), "/sink"); + + Assertions.assertThrows( + OptionValidationException.class, () -> validate(config, optionRule)); + + config.put(FileBaseOptions.DEFAULT_FS.key(), "hdfs://localhost:9000"); + Assertions.assertDoesNotThrow(() -> validate(config, optionRule)); + } + + @Test + void sinkOptionRuleRequiresFilePath() { + OptionRule optionRule = (new HdfsFileSinkFactory()).optionRule(); + Map<String, Object> config = new HashMap<>(); + config.put(FileBaseOptions.DEFAULT_FS.key(), "hdfs://localhost:9000"); + + Assertions.assertThrows( + OptionValidationException.class, () -> validate(config, optionRule)); + + config.put(FileBaseOptions.FILE_PATH.key(), "/sink"); + Assertions.assertDoesNotThrow(() -> validate(config, optionRule)); + } + private static Map<String, Object> sourceConfig() { Map<String, Object> config = new HashMap<>(); config.put(FileBaseOptions.FILE_PATH.key(), "/source"); diff --git a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/sink/S3FileSink.java b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/sink/S3FileSink.java index bca5be8368..ce183718ad 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/sink/S3FileSink.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/sink/S3FileSink.java @@ -17,9 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.file.s3.sink; -import org.apache.seatunnel.shade.com.typesafe.config.Config; - -import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode; import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.api.sink.DataSaveMode; import org.apache.seatunnel.api.sink.DefaultSaveModeHandler; @@ -29,13 +26,8 @@ import org.apache.seatunnel.api.sink.SupportSaveMode; import org.apache.seatunnel.api.table.catalog.Catalog; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.factory.CatalogFactory; -import org.apache.seatunnel.common.config.CheckConfigUtil; -import org.apache.seatunnel.common.config.CheckResult; -import org.apache.seatunnel.common.constants.PluginType; import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSinkOptions; import org.apache.seatunnel.connectors.seatunnel.file.config.FileSystemType; -import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException; -import org.apache.seatunnel.connectors.seatunnel.file.s3.config.S3FileSinkOptions; import org.apache.seatunnel.connectors.seatunnel.file.s3.config.S3HadoopConf; import org.apache.seatunnel.connectors.seatunnel.file.sink.BaseMultipleTableFileSink; @@ -59,19 +51,6 @@ public class S3FileSink extends BaseMultipleTableFileSink implements SupportSave super(S3HadoopConf.buildWithReadOnlyConfig(readonlyConfig), readonlyConfig, catalogTable); this.catalogTable = catalogTable; this.readonlyConfig = readonlyConfig; - Config pluginConfig = readonlyConfig.toConfig(); - CheckResult result = - CheckConfigUtil.checkAllExists( - pluginConfig, - S3FileSinkOptions.FILE_PATH.key(), - S3FileSinkOptions.S3_BUCKET.key()); - if (!result.isSuccess()) { - throw new FileConnectorException( - SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED, - String.format( - "PluginName: %s, PluginType: %s, Message: %s", - getPluginName(), PluginType.SINK, result.getMsg())); - } } @Override diff --git a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/S3FileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/S3FileFactoryTest.java index 756fbe7e8b..38d21352d6 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/S3FileFactoryTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/S3FileFactoryTest.java @@ -18,9 +18,12 @@ package org.apache.seatunnel.connectors.seatunnel.file.s3; import org.apache.seatunnel.api.configuration.Option; +import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.api.configuration.util.Condition; +import org.apache.seatunnel.api.configuration.util.ConfigValidator; import org.apache.seatunnel.api.configuration.util.Expression; import org.apache.seatunnel.api.configuration.util.OptionRule; +import org.apache.seatunnel.api.configuration.util.OptionValidationException; import org.apache.seatunnel.api.configuration.util.RequiredOption; import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions; import org.apache.seatunnel.connectors.seatunnel.file.s3.config.S3FileBaseOptions; @@ -30,6 +33,9 @@ import org.apache.seatunnel.connectors.seatunnel.file.s3.source.S3FileSourceFact import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.HashMap; +import java.util.Map; + class S3FileFactoryTest { @Test @@ -94,6 +100,47 @@ class S3FileFactoryTest { "S3File sink optionRule should require secret_key when the credentials provider is SimpleAWSCredentialsProvider"); } + @Test + void sinkOptionRuleRequiresFilePath() { + OptionRule optionRule = new S3FileSinkFactory().optionRule(); + Map<String, Object> config = sinkConfig(); + config.remove(S3FileBaseOptions.FILE_PATH.key()); + + Assertions.assertThrows( + OptionValidationException.class, () -> validate(config, optionRule)); + + config.put(S3FileBaseOptions.FILE_PATH.key(), "/tmp/seatunnel"); + Assertions.assertDoesNotThrow(() -> validate(config, optionRule)); + } + + @Test + void sinkOptionRuleRequiresBucket() { + OptionRule optionRule = new S3FileSinkFactory().optionRule(); + Map<String, Object> config = sinkConfig(); + config.remove(S3FileBaseOptions.S3_BUCKET.key()); + + Assertions.assertThrows( + OptionValidationException.class, () -> validate(config, optionRule)); + + config.put(S3FileBaseOptions.S3_BUCKET.key(), "s3a://seatunnel-test"); + Assertions.assertDoesNotThrow(() -> validate(config, optionRule)); + } + + private static Map<String, Object> sinkConfig() { + Map<String, Object> config = new HashMap<>(); + config.put(S3FileBaseOptions.FILE_PATH.key(), "/tmp/seatunnel"); + config.put(S3FileBaseOptions.S3_BUCKET.key(), "s3a://seatunnel-test"); + config.put(S3FileBaseOptions.FS_S3A_ENDPOINT.key(), "s3.example.com"); + config.put( + S3FileBaseOptions.S3A_AWS_CREDENTIALS_PROVIDER_CLASS.key(), + S3FileBaseOptions.INSTANCE_PROFILE_CREDENTIALS_PROVIDER); + return config; + } + + private static void validate(Map<String, Object> config, OptionRule optionRule) { + ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(optionRule); + } + private static boolean optionRuleContains(OptionRule rule, Option<?> option) { if (rule.getOptionalOptions().contains(option)) { return true;
