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 3339132550 [Feature][Connector-V2] Support S3 source connectivity
dry-run (#12242)
3339132550 is described below
commit 3339132550594fe3128534cf37c86f0cbbe435cc
Author: Goutam Adwant <[email protected]>
AuthorDate: Sun Sep 13 10:31:49 2026 +0000
[Feature][Connector-V2] Support S3 source connectivity dry-run (#12242)
Signed-off-by: Goutam Adwant <[email protected]>
---
docs/en/connectors/source/S3File.md | 40 ++
docs/en/engines/zeta/user-command.md | 1 +
docs/zh/connectors/source/S3File.md | 32 ++
docs/zh/engines/zeta/user-command.md | 1 +
.../file/s3/source/S3FileSourceFactory.java | 19 +-
.../file/s3/source/S3SourceDryRunValidator.java | 284 +++++++++++++
.../seatunnel/file/s3/S3FileSourceFactoryTest.java | 8 +
.../s3/source/S3SourceDryRunValidatorTest.java | 463 +++++++++++++++++++++
.../connector/file/s3/S3FileConnectDryRunIT.java | 190 +++++++++
.../file/s3/S3FileConnectDryRunWireIT.java | 249 +++++++++++
10 files changed, 1286 insertions(+), 1 deletion(-)
diff --git a/docs/en/connectors/source/S3File.md
b/docs/en/connectors/source/S3File.md
index 37ec943dd6..5672df90fb 100644
--- a/docs/en/connectors/source/S3File.md
+++ b/docs/en/connectors/source/S3File.md
@@ -41,6 +41,46 @@ import ChangeLog from '../changelog/connector-file-s3.md';
Read data from aws s3 file system.
+### Connectivity dry-run
+
+`--dry-run connect` can check a single S3A source before submitting a job. The
+source must use `bucket = "s3a://your-bucket"`, an absolute `path`, an explicit
+inline `schema.fields` or `schema.columns`, and `file_format_type` of `text`,
+`csv`, `json`, or `xml`. Set `parse_partition_from_path = false` and omit
+`read_columns`; file-derived schemas, projection, and partition inference are
+not validated by this metadata-only check. Unsupported configurations fail the
+connect dry-run with an explanation; normal job execution is unchanged.
+
+The check reuses Hadoop S3A endpoint, credential-chain, proxy and path-style
+configuration. It checks object metadata with HEAD, or makes one prefix listing
+with `maxKeys=1` and delimiter `/`. It does not open file contents, recursively
+list files, create readers, upload, delete, or initialize a shared filesystem.
+An exact object does not require listing permission. A prefix requires listing
+permission; a successful empty listing is accepted for `discovery_mode =
+"continuous"`, since files may arrive later. For a batch source, an empty
+virtual prefix without a directory marker fails; an accessible empty bucket
+root is accepted. Missing buckets and denied requests fail in both modes.
+
+Validation-only connection establishment and socket timeouts are capped at
+5 seconds, preserving smaller positive values. SDK request retries are disabled
+for validation, including bucket-specific overrides. These are network timeout
+settings, not a total deadline for DNS, credential-provider initialization, or
+SDK setup. Runtime timeouts and retries are unchanged.
+
+This initial check does not support `tables_configs`, legacy `s3n` buckets,
+SSE-C customer-provided encryption keys,
`fs.s3a.security.credential.provider.path`,
+S3Guard, multipart purge, or custom S3 client factories. It does not prove
object
+content readability, file-format correctness, schema compatibility with the
+stored data, worker-side credentials, or target/update/post-sync permissions.
+
+Save your job configuration meeting the requirements above as
+`config/s3-to-console.conf` (this is a user-created file, not a bundled
template),
+then run from the SeaTunnel installation directory:
+
+```bash
+bin/seatunnel.sh --config config/s3-to-console.conf --dry-run connect -e local
+```
+
## Supported DataSource Info
| Datasource | Supported versions |
diff --git a/docs/en/engines/zeta/user-command.md
b/docs/en/engines/zeta/user-command.md
index 433f5517af..cbe38044e6 100644
--- a/docs/en/engines/zeta/user-command.md
+++ b/docs/en/engines/zeta/user-command.md
@@ -83,6 +83,7 @@ The `--dry-run connect` option runs the static checks first,
then uses connector
| Jdbc | Yes (connectivity + schema inference) | Yes (connectivity +
table existence + field compatibility) |
| Kafka | Yes ([topic metadata + runtime output
schema](../../connectors/source/Kafka.md#connectivity-dry-run), not
consumer/group permissions) | No |
| FakeSource | Yes (schema inference only, no external system) | - |
+| S3File | Yes (metadata connectivity + inline schema, single-table
text/csv/json/xml; see [supported
scope](../../connectors/source/S3File.md#connectivity-dry-run)) | - |
Every plugin in the job is reported in a validation summary with one of two
statuses:
diff --git a/docs/zh/connectors/source/S3File.md
b/docs/zh/connectors/source/S3File.md
index 479ffe6418..4e9e0a4011 100644
--- a/docs/zh/connectors/source/S3File.md
+++ b/docs/zh/connectors/source/S3File.md
@@ -41,6 +41,38 @@ import ChangeLog from '../changelog/connector-file-s3.md';
从aws s3文件系统读取数据。
+### 连接预检查
+
+`--dry-run connect` 可以在提交作业前检查单表 S3A 数据源。需要配置
+`bucket = "s3a://your-bucket"`、绝对 `path`、显式内联 `schema.fields` 或
+`schema.columns`，并将 `file_format_type` 设置为 `text`、`csv`、`json` 或
+`xml`。同时设置 `parse_partition_from_path = false`，且不配置 `read_columns`。
+此元数据检查不推断文件结构、投影或分区字段。不支持的配置会使连接预检查失败并
+返回说明，不改变正常作业的行为。
+
+检查复用 Hadoop S3A 的端点、凭证链、代理和路径访问配置，使用 HEAD 获取对象元数据，
+或执行一次 `maxKeys=1`、分隔符为 `/` 的前缀列表请求。它不会打开文件内容、递归列举
+文件、创建读取器、上传或删除对象，也不会初始化共享文件系统。检查具体对象不需要
+列举权限，检查前缀需要列举权限。对于 `discovery_mode = "continuous"`，成功的空列表
+可以通过检查，因为文件可能稍后到达。批处理模式下，没有目录标记的空虚拟前缀会失败，
+但可访问的空桶根路径可以通过。桶不存在或请求被拒绝时，两种模式均失败。
+
+预检查将建立连接和套接字超时限制为最多 5 秒，并保留更小的正值，禁用 SDK 请求重试，
+这些限制也适用于桶级覆盖配置。它们不是 DNS、凭证提供器初始化或 SDK 初始化的总耗时
+限制，不改变正常作业的超时和重试设置。
+
+首个版本不支持 `tables_configs`、旧 `s3n` 桶、SSE-C 客户提供的加密密钥、
+`fs.s3a.security.credential.provider.path`、S3Guard、分段上传清理以及自定义 S3 客户端工厂。
+通过检查不代表文件内容可读、格式或数据结构正确，也不验证工作节点凭证、目标端、更新
+或同步后操作的权限。
+
+将满足上述条件的作业配置保存为 `config/s3-to-console.conf`（这是用户自行创建的文件，
+并非内置模板），然后在 SeaTunnel 安装目录中执行：
+
+```bash
+bin/seatunnel.sh --config config/s3-to-console.conf --dry-run connect -e local
+```
+
## 支持的数据源信息
| 数据源 | 支持的版本 |
diff --git a/docs/zh/engines/zeta/user-command.md
b/docs/zh/engines/zeta/user-command.md
index cded907c75..d708124799 100644
--- a/docs/zh/engines/zeta/user-command.md
+++ b/docs/zh/engines/zeta/user-command.md
@@ -99,6 +99,7 @@ bin/seatunnel.sh --config
$SEATUNNEL_HOME/config/v2.batch.config.template --dry-
| Jdbc | 支持（连通性 + schema 推断） | 支持（连通性 + 表存在性 + 字段兼容性） |
| Kafka | 支持（[主题元数据 + 运行时输出
schema](../../connectors/source/Kafka.md#连通性-dry-run)，不含消费或消费组权限） | 不支持 |
| FakeSource | 支持（仅 schema 推断，无外部系统） | - |
+| S3File | 支持（元数据连通性 + 内联 schema，仅单表
text/csv/json/xml；参见[支持范围](../../connectors/source/S3File.md#连接预检查)） | - |
作业中的每个插件都会在校验汇总中报告以下两种状态之一：
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java
index 083d6ee211..02a85e7cd4 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java
@@ -21,8 +21,10 @@ import
org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.options.ConnectorCommonOptions;
import org.apache.seatunnel.api.source.SeaTunnelSource;
import org.apache.seatunnel.api.source.SourceSplit;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.connector.TableSource;
import org.apache.seatunnel.api.table.factory.Factory;
+import org.apache.seatunnel.api.table.factory.SupportSourceDryRunValidation;
import org.apache.seatunnel.api.table.factory.TableSourceFactory;
import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
import
org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
@@ -36,9 +38,24 @@ import com.google.auto.service.AutoService;
import java.io.Serializable;
import java.util.Arrays;
+import java.util.List;
@AutoService(Factory.class)
-public class S3FileSourceFactory implements TableSourceFactory {
+public class S3FileSourceFactory implements TableSourceFactory,
SupportSourceDryRunValidation {
+ /** Returns the configured schema without discovering or opening source
files. */
+ @Override
+ public List<CatalogTable> inferSchemaForDryRun(TableSourceFactoryContext
context) {
+ S3SourceDryRunValidator.validateSchemaOptions(context.getOptions());
+ return discoverTableSchemas(context);
+ }
+
+ /** Checks S3A path metadata using an independently owned client. */
+ @Override
+ public void validateConnectionForDryRun(
+ TableSourceFactoryContext context, List<CatalogTable>
catalogTables) throws Exception {
+ S3SourceDryRunValidator.validate(context.getOptions());
+ }
+
@Override
public String factoryIdentifier() {
return FileSystemType.S3.getFileSystemPluginName();
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3SourceDryRunValidator.java
b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3SourceDryRunValidator.java
new file mode 100644
index 0000000000..ec1da6dd5b
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3SourceDryRunValidator.java
@@ -0,0 +1,284 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.file.s3.source;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.options.ConnectorCommonOptions;
+import org.apache.seatunnel.api.options.table.ColumnOptions;
+import org.apache.seatunnel.api.options.table.FieldOptions;
+import
org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
+import org.apache.seatunnel.connectors.seatunnel.file.config.FileDiscoveryMode;
+import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
+import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
+import
org.apache.seatunnel.connectors.seatunnel.file.s3.config.S3FileSourceOptions;
+import org.apache.seatunnel.connectors.seatunnel.file.s3.config.S3HadoopConf;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.s3a.AWSCredentialProviderList;
+import org.apache.hadoop.fs.s3a.Constants;
+import org.apache.hadoop.fs.s3a.DefaultS3ClientFactory;
+import org.apache.hadoop.fs.s3a.S3AUtils;
+
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.model.AmazonS3Exception;
+import com.amazonaws.services.s3.model.ListObjectsRequest;
+import com.amazonaws.services.s3.model.ListObjectsV2Request;
+import com.amazonaws.services.s3.model.ListObjectsV2Result;
+import com.amazonaws.services.s3.model.ObjectListing;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Arrays;
+import java.util.Map;
+
+final class S3SourceDryRunValidator {
+ private static final int NETWORK_TIMEOUT_MILLIS = 5000;
+
+ private S3SourceDryRunValidator() {}
+
+ /** Rejects schemas that require opening files or altering the configured
field layout. */
+ static void validateSchemaOptions(ReadonlyConfig options) {
+ if
(options.getOptional(ConnectorCommonOptions.TABLE_CONFIGS).isPresent()
+ ||
!options.getOptional(ConnectorCommonOptions.SCHEMA).isPresent()) {
+ throw new IllegalArgumentException(
+ "S3File connect dry-run requires a single table with an
explicit inline schema.");
+ }
+ Map<String, Object> schema =
options.get(ConnectorCommonOptions.SCHEMA);
+ if (schema.containsKey(ColumnOptions.METADATA_TABLE_ID.key())
+ || (!options.getOptional(FieldOptions.FIELDS).isPresent()
+ && !schema.containsKey(ColumnOptions.COLUMNS.key()))) {
+ throw new IllegalArgumentException(
+ "S3File connect dry-run requires schema.fields or
schema.columns, not metadata schema lookup.");
+ }
+ if (!Arrays.asList(FileFormat.TEXT, FileFormat.CSV, FileFormat.JSON,
FileFormat.XML)
+
.contains(options.get(FileBaseSourceOptions.FILE_FORMAT_TYPE))) {
+ throw new IllegalArgumentException(
+ "S3File connect dry-run supports text, csv, json and xml
without reading file contents.");
+ }
+ if (options.get(FileBaseSourceOptions.PARSE_PARTITION_FROM_PATH)
+ ||
options.getOptional(FileBaseSourceOptions.READ_COLUMNS).isPresent()) {
+ throw new IllegalArgumentException(
+ "S3File connect dry-run requires
parse_partition_from_path=false and no read_columns to preserve the runtime
schema.");
+ }
+ }
+
+ /** Checks metadata with a private client, closing credentials even when
client setup fails. */
+ static void validate(ReadonlyConfig options) throws IOException {
+ // The connection hook can be called directly, without the preceding
schema hook.
+ validateSchemaOptions(options);
+ URI bucket = bucketUri(options);
+ Path path = sourcePath(options, bucket);
+ Configuration configuration = validationConfiguration(options, bucket);
+ // Validate pure SDK settings before Hadoop creates any credential
providers.
+ DefaultS3ClientFactory.createAwsConf(configuration);
+ try (DryRunClientFactory factory = new DryRunClientFactory()) {
+ factory.setConf(configuration);
+ AmazonS3 client = factory.createS3Client(bucket);
+ validatePath(
+ client,
+ bucket.getHost(),
+ path,
+ options.get(FileBaseSourceOptions.DISCOVERY_MODE),
+ configuration.getInt(Constants.LIST_VERSION,
Constants.DEFAULT_LIST_VERSION));
+ }
+ }
+
+ /** Accepts only a bare S3A bucket URI, without echoing potentially
sensitive input. */
+ static URI bucketUri(ReadonlyConfig options) {
+ URI bucket;
+ try {
+ bucket = URI.create(options.get(S3FileSourceOptions.S3_BUCKET));
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ "S3File connect dry-run requires an s3a://bucket URI.");
+ }
+ if (!"s3a".equals(bucket.getScheme())
+ || bucket.getHost() == null
+ || bucket.getUserInfo() != null
+ || bucket.getPort() != -1
+ || bucket.getQuery() != null
+ || bucket.getFragment() != null
+ || (!bucket.getPath().isEmpty() &&
!"/".equals(bucket.getPath()))) {
+ throw new IllegalArgumentException(
+ "S3File connect dry-run requires an s3a://bucket URI.");
+ }
+ return bucket;
+ }
+
+ /** Keeps metadata requests inside the configured bucket, using the
source's absolute path. */
+ static Path sourcePath(ReadonlyConfig options, URI bucket) {
+ Path path;
+ try {
+ path = new Path(options.get(FileBaseSourceOptions.FILE_PATH));
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ "S3File connect dry-run requires an absolute path in the
configured bucket.");
+ }
+ URI uri = path.toUri();
+ if (!path.isAbsolute()
+ || (uri.getScheme() != null && !"s3a".equals(uri.getScheme()))
+ || (uri.getAuthority() != null
+ && !bucket.getAuthority().equals(uri.getAuthority())))
{
+ throw new IllegalArgumentException(
+ "S3File connect dry-run requires an absolute path in the
configured bucket.");
+ }
+ return path;
+ }
+
+ /**
+ * Preserves S3A connection settings but rejects filesystem side effects,
custom client
+ * lifecycle behavior, and encryption/credential-store paths not supported
by this metadata-only
+ * client. Disables retries so a preflight check does not inherit the
production retry budget.
+ */
+ static Configuration validationConfiguration(ReadonlyConfig options, URI
bucket) {
+ HadoopConf hadoopConf = S3HadoopConf.buildWithReadOnlyConfig(options);
+ Configuration configuration = hadoopConf.toConfiguration();
+ hadoopConf.setExtraOptionsForConfiguration(configuration);
+ configuration = S3AUtils.propagateBucketOptions(configuration,
bucket.getHost());
+ if (configuration.getBoolean(Constants.PURGE_EXISTING_MULTIPART, false)
+ || !Constants.S3GUARD_METASTORE_NULL.equals(
+ configuration.get(
+ Constants.S3_METADATA_STORE_IMPL,
Constants.S3GUARD_METASTORE_NULL))
+ || !Constants.DEFAULT_S3_CLIENT_FACTORY_IMPL
+ .getName()
+ .equals(
+ configuration.get(
+ Constants.S3_CLIENT_FACTORY_IMPL,
+
Constants.DEFAULT_S3_CLIENT_FACTORY_IMPL.getName()))) {
+ throw new IllegalArgumentException(
+ "S3File connect dry-run does not support multipart purge,
S3Guard or a custom S3 client factory.");
+ }
+ if
(!configuration.getTrimmed(Constants.S3A_SECURITY_CREDENTIAL_PROVIDER_PATH,
"").isEmpty()
+ || "SSE-C"
+ .equalsIgnoreCase(
+ configuration.getTrimmed(
+
Constants.SERVER_SIDE_ENCRYPTION_ALGORITHM, ""))) {
+ throw new IllegalArgumentException(
+ "S3File connect dry-run does not support SSE-C or
fs.s3a.security.credential.provider.path.");
+ }
+ clampTimeout(configuration, bucket, Constants.ESTABLISH_TIMEOUT);
+ clampTimeout(configuration, bucket, Constants.SOCKET_TIMEOUT);
+ for (String key :
+ Arrays.asList(
+ Constants.MAX_ERROR_RETRIES,
+ Constants.RETRY_LIMIT,
+ Constants.RETRY_THROTTLE_LIMIT)) {
+ S3AUtils.clearBucketOption(configuration, bucket.getHost(), key);
+ configuration.setInt(key, 0);
+ }
+ return configuration;
+ }
+
+ /** Caps each network wait at five seconds for preflight, retaining
stricter user limits. */
+ private static void clampTimeout(Configuration configuration, URI bucket,
String key) {
+ int configured = configuration.getInt(key, NETWORK_TIMEOUT_MILLIS);
+ if (configured < 0) {
+ throw new IllegalArgumentException(
+ "S3File connect dry-run requires non-negative " + key);
+ }
+ S3AUtils.clearBucketOption(configuration, bucket.getHost(), key);
+ configuration.setInt(
+ key,
+ configured == 0
+ ? NETWORK_TIMEOUT_MILLIS
+ : Math.min(configured, NETWORK_TIMEOUT_MILLIS));
+ }
+
+ /**
+ * Tries HEAD first so exact objects need no listing permission. Only a
missing object falls
+ * back to one bounded prefix listing; permission and other failures must
not be treated as
+ * absence.
+ */
+ static void validatePath(
+ AmazonS3 client,
+ String bucket,
+ Path path,
+ FileDiscoveryMode discoveryMode,
+ int listVersion)
+ throws IOException {
+ String key = path.toUri().getPath().substring(1);
+ if (!key.isEmpty()) {
+ try {
+ client.getObjectMetadata(bucket, key);
+ return;
+ } catch (AmazonS3Exception e) {
+ if (e.getStatusCode() != 404) {
+ throw e;
+ }
+ }
+ }
+ String prefix = key.isEmpty() || key.endsWith("/") ? key : key + "/";
+ boolean empty;
+ if (listVersion == 1) {
+ ObjectListing result =
+ client.listObjects(
+ new ListObjectsRequest()
+ .withBucketName(bucket)
+ .withPrefix(prefix)
+ .withDelimiter("/")
+ .withMaxKeys(1));
+ empty = result.getObjectSummaries().isEmpty() &&
result.getCommonPrefixes().isEmpty();
+ } else {
+ ListObjectsV2Result result =
+ client.listObjectsV2(
+ new ListObjectsV2Request()
+ .withBucketName(bucket)
+ .withPrefix(prefix)
+ .withDelimiter("/")
+ .withMaxKeys(1));
+ empty = result.getObjectSummaries().isEmpty() &&
result.getCommonPrefixes().isEmpty();
+ }
+ // Only a successful empty listing proves an accessible prefix
awaiting its first file.
+ if (empty && !key.isEmpty() && discoveryMode !=
FileDiscoveryMode.CONTINUOUS) {
+ throw new FileNotFoundException(
+ "S3File connect dry-run could not find the configured
batch source path.");
+ }
+ }
+
+ /** Owns the client and credential chain without creating a shared S3A
filesystem. */
+ private static final class DryRunClientFactory extends
DefaultS3ClientFactory
+ implements AutoCloseable {
+ private AmazonS3 client;
+ private AWSCredentialProviderList credentials;
+
+ @Override
+ protected AmazonS3 newAmazonS3Client(
+ AWSCredentialsProvider credentials, ClientConfiguration
configuration) {
+ this.credentials = (AWSCredentialProviderList) credentials;
+ this.client = super.newAmazonS3Client(credentials, configuration);
+ return client;
+ }
+
+ @Override
+ public void close() {
+ try {
+ if (client != null) {
+ client.shutdown();
+ }
+ } finally {
+ if (credentials != null) {
+ credentials.close();
+ }
+ }
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/S3FileSourceFactoryTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/S3FileSourceFactoryTest.java
index a60221fbc2..c222b7d19a 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/S3FileSourceFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/S3FileSourceFactoryTest.java
@@ -21,6 +21,7 @@ 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.api.table.factory.SupportSourceDryRunValidation;
import
org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.file.s3.config.S3FileSourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.file.s3.source.S3FileSourceFactory;
@@ -36,6 +37,13 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
class S3FileSourceFactoryTest {
+ @Test
+ void shouldSupportConnectivityDryRun() {
+ assertTrue(
+ new S3FileSourceFactory() instanceof
SupportSourceDryRunValidation,
+ "S3File source connectivity is skipped when the factory does
not expose the dry-run SPI");
+ }
+
@Test
void shouldExposeContinuousDiscoveryOptions() {
OptionRule optionRule = new S3FileSourceFactory().optionRule();
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3SourceDryRunValidatorTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3SourceDryRunValidatorTest.java
new file mode 100644
index 0000000000..5be35b9230
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3SourceDryRunValidatorTest.java
@@ -0,0 +1,463 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.file.s3.source;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
+import org.apache.seatunnel.connectors.seatunnel.file.config.FileDiscoveryMode;
+import
org.apache.seatunnel.connectors.seatunnel.file.s3.config.S3FileSourceOptions;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.s3a.Constants;
+import org.apache.hadoop.fs.s3a.S3AUtils;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.model.AmazonS3Exception;
+import com.amazonaws.services.s3.model.ListObjectsRequest;
+import com.amazonaws.services.s3.model.ListObjectsV2Request;
+import com.amazonaws.services.s3.model.ListObjectsV2Result;
+import com.amazonaws.services.s3.model.ObjectListing;
+import com.amazonaws.services.s3.model.ObjectMetadata;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+class S3SourceDryRunValidatorTest {
+ private static final URI BUCKET = URI.create("s3a://warehouse");
+
+ @ParameterizedTest
+ @ValueSource(strings = {"text", "csv", "json", "xml"})
+ void shouldReturnConfiguredSchemaWithoutCreatingSource(String format) {
+ Map<String, Object> config = sourceConfig();
+ config.put("file_format_type", format);
+ TableSourceFactoryContext context = context(config);
+ S3FileSourceFactory factory = new S3FileSourceFactory();
+
+ List<CatalogTable> tables = factory.inferSchemaForDryRun(context);
+
+ assertEquals(1, tables.size());
+ assertEquals(
+
factory.discoverTableSchemas(context).get(0).getSeaTunnelRowType(),
+ tables.get(0).getSeaTunnelRowType());
+ assertEquals("event_id",
tables.get(0).getSeaTunnelRowType().getFieldName(0));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"binary", "markdown", "pdf", "parquet", "orc",
"excel"})
+ void shouldRefuseFormatsRequiringDifferentSchemaHandling(String format) {
+ Map<String, Object> config = sourceConfig();
+ config.put("file_format_type", format);
+ assertSchemaFailure(config, "supports text, csv, json and xml");
+ }
+
+ @Test
+ void shouldRefuseMissingSchema() {
+ Map<String, Object> config = sourceConfig();
+ config.remove("schema");
+ assertSchemaFailure(config, "explicit inline schema");
+ }
+
+ @Test
+ void shouldRefuseMetadataSchemaLookup() {
+ Map<String, Object> config = sourceConfig();
+ config.put("schema", Collections.singletonMap("metadata_table_id",
"warehouse.events"));
+ assertSchemaFailure(config, "not metadata schema lookup");
+ }
+
+ @Test
+ void shouldRefuseTableConfigs() {
+ Map<String, Object> config = sourceConfig();
+ config.put("tables_configs",
Collections.singletonList(sourceConfig()));
+ assertSchemaFailure(config, "single table");
+ }
+
+ @Test
+ void shouldRefuseDefaultPartitionInference() {
+ Map<String, Object> config = sourceConfig();
+ config.remove("parse_partition_from_path");
+ assertSchemaFailure(config, "parse_partition_from_path=false");
+ }
+
+ @Test
+ void shouldRefuseProjection() {
+ Map<String, Object> config = sourceConfig();
+ config.put("read_columns", Collections.singletonList("event_id"));
+ assertSchemaFailure(config, "no read_columns");
+ }
+
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "s3n://warehouse",
+ "s3a://warehouse/folder",
+ "s3a://secret@warehouse",
+ "s3a://warehouse?key=secret"
+ })
+ void shouldRefuseUnsupportedBucketUriWithoutEchoingIt(String bucket) {
+ Map<String, Object> config = sourceConfig();
+ config.put("bucket", bucket);
+ IllegalArgumentException failure =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
S3SourceDryRunValidator.bucketUri(ReadonlyConfig.fromMap(config)));
+ assertTrue(failure.getMessage().contains("s3a://bucket URI"));
+ assertFalse(failure.getMessage().contains("secret"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "events",
+ "s3a://other/events",
+ "file:///events",
+ "s3a://secret@warehouse/events",
+ "s3a://[secret"
+ })
+ void shouldRefusePathOutsideConfiguredBucket(String path) {
+ Map<String, Object> config = sourceConfig();
+ config.put("path", path);
+ IllegalArgumentException failure =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ S3SourceDryRunValidator.sourcePath(
+ ReadonlyConfig.fromMap(config),
BUCKET));
+ assertTrue(failure.getMessage().contains("absolute path in the
configured bucket"));
+ assertFalse(failure.getMessage().contains("secret"));
+ }
+
+ @Test
+ void shouldRetainEndpointAndCredentialsWithoutChangingInputProperties() {
+ Map<String, Object> config = sourceConfig();
+ Map<String, String> properties = new HashMap<>();
+ properties.put(Constants.PATH_STYLE_ACCESS, "true");
+ properties.put(Constants.SOCKET_TIMEOUT, "1200");
+ config.put("hadoop_s3_properties", properties);
+
+ Configuration configuration = configuration(config);
+
+ assertEquals("http://localhost:9000";,
configuration.get(Constants.ENDPOINT));
+ assertEquals("access-key", configuration.get(Constants.ACCESS_KEY));
+ assertEquals("secret-key", configuration.get(Constants.SECRET_KEY));
+ assertEquals(
+ S3FileSourceOptions.SIMPLE_AWS_CREDENTIALS_PROVIDER,
+ configuration.get(Constants.AWS_CREDENTIALS_PROVIDER));
+ assertEquals(1200, configuration.getInt(Constants.SOCKET_TIMEOUT, -1));
+ assertTrue(configuration.getBoolean(Constants.PATH_STYLE_ACCESS,
false));
+ assertEquals(2, properties.size());
+ assertEquals("1200", properties.get(Constants.SOCKET_TIMEOUT));
+ }
+
+ @Test
+ void
shouldBoundBucketSpecificTimeoutsAndRetriesAfterS3AReappliesOverrides() {
+ Map<String, Object> config = sourceConfig();
+ Map<String, String> properties = new HashMap<>();
+ properties.put("fs.s3a.bucket.warehouse.connection.timeout", "0");
+ properties.put("fs.s3a.bucket.warehouse.connection.establish.timeout",
"2000");
+ properties.put("fs.s3a.bucket.warehouse.attempts.maximum", "99");
+ properties.put("fs.s3a.bucket.warehouse.retry.limit", "99");
+ properties.put("fs.s3a.bucket.warehouse.retry.throttle.limit", "99");
+ config.put("hadoop_s3_properties", properties);
+
+ Configuration configuration =
+ S3AUtils.propagateBucketOptions(configuration(config),
"warehouse");
+
+ assertEquals(5000, configuration.getInt(Constants.SOCKET_TIMEOUT, -1));
+ assertEquals(2000, configuration.getInt(Constants.ESTABLISH_TIMEOUT,
-1));
+ assertEquals(0, configuration.getInt(Constants.MAX_ERROR_RETRIES, -1));
+ assertEquals(0, configuration.getInt(Constants.RETRY_LIMIT, -1));
+ assertEquals(0, configuration.getInt(Constants.RETRY_THROTTLE_LIMIT,
-1));
+ assertEquals("99",
properties.get("fs.s3a.bucket.warehouse.retry.limit"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"fs.s3a.multipart.purge",
"fs.s3a.bucket.warehouse.multipart.purge"})
+ void shouldRefuseMultipartPurgeBeforeOpeningFilesystem(String key) {
+ assertConfigurationFailure(key, "true", "multipart purge");
+ }
+
+ @ParameterizedTest
+ @ValueSource(
+ strings = {"fs.s3a.metadatastore.impl",
"fs.s3a.bucket.warehouse.metadatastore.impl"})
+ void shouldRefuseMetadataStoreInitialization(String key) {
+ assertConfigurationFailure(key, Constants.S3GUARD_METASTORE_DYNAMO,
"S3Guard");
+ }
+
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ "fs.s3a.s3.client.factory.impl",
+ "fs.s3a.bucket.warehouse.s3.client.factory.impl"
+ })
+ void shouldRefuseCustomClientInitialization(String key) {
+ assertConfigurationFailure(key, "example.CustomClientFactory", "custom
S3 client factory");
+ }
+
+ @Test
+ void shouldRefuseNegativeTimeout() {
+ assertConfigurationFailure(Constants.SOCKET_TIMEOUT, "-1",
"non-negative");
+ }
+
+ @Test
+ void shouldCheckOnlyPathMetadata() throws IOException {
+ AmazonS3 client = mock(AmazonS3.class);
+ Path path = new Path("/events");
+ when(client.getObjectMetadata("warehouse", "events")).thenReturn(new
ObjectMetadata());
+ S3SourceDryRunValidator.validatePath(client, "warehouse", path,
FileDiscoveryMode.ONCE, 2);
+ verify(client).getObjectMetadata("warehouse", "events");
+ verifyNoMoreInteractions(client);
+ }
+
+ @Test
+ void shouldFailForMissingBatchPath() throws IOException {
+ AmazonS3 client = emptyPrefixClient();
+ Path path = new Path("/events");
+ FileNotFoundException failure =
+ assertThrows(
+ FileNotFoundException.class,
+ () ->
+ S3SourceDryRunValidator.validatePath(
+ client, "warehouse", path,
FileDiscoveryMode.ONCE, 2));
+ assertTrue(failure.getMessage().contains("batch source path"));
+ }
+
+ @Test
+ void shouldAllowEmptyContinuousPrefix() throws IOException {
+ AmazonS3 client = emptyPrefixClient();
+ Path path = new Path("/events");
+ assertDoesNotThrow(
+ () ->
+ S3SourceDryRunValidator.validatePath(
+ client, "warehouse", path,
FileDiscoveryMode.CONTINUOUS, 2));
+ }
+
+ @Test
+ void shouldPropagateDeniedAccessForContinuousDiscovery() throws
IOException {
+ AmazonS3 client = emptyPrefixClient();
+ Path path = new Path("/events");
+ AmazonS3Exception denied = serviceFailure(403);
+
when(client.listObjectsV2(any(ListObjectsV2Request.class))).thenThrow(denied);
+ AmazonS3Exception failure =
+ assertThrows(
+ AmazonS3Exception.class,
+ () ->
+ S3SourceDryRunValidator.validatePath(
+ client,
+ "warehouse",
+ path,
+ FileDiscoveryMode.CONTINUOUS,
+ 2));
+ assertSame(denied, failure);
+ assertTrue(failure.getMessage().contains("metadata failure"));
+ }
+
+ @Test
+ void shouldFailForMissingBucketDuringContinuousDiscovery() {
+ AmazonS3 client = emptyPrefixClient();
+ AmazonS3Exception missing = serviceFailure(404);
+
when(client.listObjectsV2(any(ListObjectsV2Request.class))).thenThrow(missing);
+ AmazonS3Exception failure =
+ assertThrows(
+ AmazonS3Exception.class,
+ () ->
+ S3SourceDryRunValidator.validatePath(
+ client,
+ "warehouse",
+ new Path("/events"),
+ FileDiscoveryMode.CONTINUOUS,
+ 2));
+ assertSame(missing, failure);
+ assertEquals(404, failure.getStatusCode());
+ }
+
+ @Test
+ void shouldBoundPrefixListingWithoutFollowingContinuation() throws
IOException {
+ AmazonS3 client = emptyPrefixClient();
+ when(client.listObjectsV2(any(ListObjectsV2Request.class)))
+ .thenAnswer(
+ invocation -> {
+ ListObjectsV2Request request =
invocation.getArgument(0);
+ assertEquals("warehouse", request.getBucketName());
+ assertEquals("events/", request.getPrefix());
+ assertEquals("/", request.getDelimiter());
+ assertEquals(1, request.getMaxKeys());
+ ListObjectsV2Result result = new
ListObjectsV2Result();
+ result.getObjectSummaries().add(new
S3ObjectSummary());
+ result.setTruncated(true);
+ result.setNextContinuationToken("more-events");
+ return result;
+ });
+ S3SourceDryRunValidator.validatePath(
+ client, "warehouse", new Path("/events"),
FileDiscoveryMode.ONCE, 2);
+ verify(client).getObjectMetadata("warehouse", "events");
+ verify(client).listObjectsV2(any(ListObjectsV2Request.class));
+ verifyNoMoreInteractions(client);
+ }
+
+ @Test
+ void shouldRefuseCustomerProvidedEncryptionKeys() {
+ assertConfigurationFailure(Constants.SERVER_SIDE_ENCRYPTION_ALGORITHM,
"SSE-C", "SSE-C");
+ }
+
+ @Test
+ void shouldHonorLegacyListVersion() throws IOException {
+ AmazonS3 client = mock(AmazonS3.class);
+ when(client.getObjectMetadata("warehouse",
"events")).thenThrow(serviceFailure(404));
+ when(client.listObjects(any(ListObjectsRequest.class)))
+ .thenAnswer(
+ invocation -> {
+ ListObjectsRequest request =
invocation.getArgument(0);
+ assertEquals(Integer.valueOf(1),
request.getMaxKeys());
+ assertEquals("events/", request.getPrefix());
+ assertEquals("/", request.getDelimiter());
+ return new ObjectListing();
+ });
+ S3SourceDryRunValidator.validatePath(
+ client, "warehouse", new Path("/events"),
FileDiscoveryMode.CONTINUOUS, 1);
+ verify(client).getObjectMetadata("warehouse", "events");
+ verify(client).listObjects(any(ListObjectsRequest.class));
+ verifyNoMoreInteractions(client);
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {0, 3})
+ void shouldMatchS3AFallbackForOtherListVersions(int listVersion) throws
IOException {
+ AmazonS3 client = emptyPrefixClient();
+ S3SourceDryRunValidator.validatePath(
+ client,
+ "warehouse",
+ new Path("/events"),
+ FileDiscoveryMode.CONTINUOUS,
+ listVersion);
+ verify(client).getObjectMetadata("warehouse", "events");
+ verify(client).listObjectsV2(any(ListObjectsV2Request.class));
+ verifyNoMoreInteractions(client);
+ }
+
+ @Test
+ void shouldRefuseS3SpecificCredentialStoreBeforeClientInitialization() {
+ assertConfigurationFailure(
+ Constants.S3A_SECURITY_CREDENTIAL_PROVIDER_PATH,
+ "jceks://file/credentials",
+ "credential.provider.path");
+ }
+
+ private static AmazonS3 emptyPrefixClient() {
+ AmazonS3 client = mock(AmazonS3.class);
+ when(client.getObjectMetadata("warehouse",
"events")).thenThrow(serviceFailure(404));
+ when(client.listObjectsV2(any(ListObjectsV2Request.class)))
+ .thenReturn(new ListObjectsV2Result());
+ return client;
+ }
+
+ @Test
+ void shouldReturnExplicitColumnSchema() {
+ Map<String, Object> config = sourceConfig();
+ Map<String, Object> column = new HashMap<>();
+ column.put("name", "event_id");
+ column.put("type", "bigint");
+ config.put(
+ "schema", Collections.singletonMap("columns",
Collections.singletonList(column)));
+ CatalogTable table = new
S3FileSourceFactory().inferSchemaForDryRun(context(config)).get(0);
+ assertEquals("event_id", table.getSeaTunnelRowType().getFieldName(0));
+ }
+
+ @Test
+ void shouldRetainDefaultCredentialProvider() {
+ Map<String, Object> config = sourceConfig();
+ config.remove("fs.s3a.aws.credentials.provider");
+ Configuration configuration = configuration(config);
+ assertEquals(
+ S3FileSourceOptions.INSTANCE_PROFILE_CREDENTIALS_PROVIDER,
+ configuration.get(Constants.AWS_CREDENTIALS_PROVIDER));
+ }
+
+ private static AmazonS3Exception serviceFailure(int status) {
+ AmazonS3Exception failure = new AmazonS3Exception("metadata failure");
+ failure.setStatusCode(status);
+ return failure;
+ }
+
+ static Map<String, Object> sourceConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("path", "/events");
+ config.put("file_format_type", "json");
+ config.put("bucket", BUCKET.toString());
+ config.put("fs.s3a.endpoint", "http://localhost:9000";);
+ config.put(
+ "fs.s3a.aws.credentials.provider",
+ S3FileSourceOptions.SIMPLE_AWS_CREDENTIALS_PROVIDER);
+ config.put("access_key", "access-key");
+ config.put("secret_key", "secret-key");
+ config.put("parse_partition_from_path", false);
+ config.put(
+ "schema",
+ Collections.singletonMap("fields",
Collections.singletonMap("event_id", "bigint")));
+ return config;
+ }
+
+ private static TableSourceFactoryContext context(Map<String, Object>
config) {
+ return new TableSourceFactoryContext(
+ ReadonlyConfig.fromMap(config),
S3SourceDryRunValidatorTest.class.getClassLoader());
+ }
+
+ private static Configuration configuration(Map<String, Object> config) {
+ return S3SourceDryRunValidator.validationConfiguration(
+ ReadonlyConfig.fromMap(config), BUCKET);
+ }
+
+ private static void assertSchemaFailure(Map<String, Object> config, String
message) {
+ IllegalArgumentException failure =
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> new
S3FileSourceFactory().inferSchemaForDryRun(context(config)));
+ assertTrue(failure.getMessage().contains(message),
failure.getMessage());
+ }
+
+ private static void assertConfigurationFailure(String key, String value,
String message) {
+ Map<String, Object> config = sourceConfig();
+ config.put("hadoop_s3_properties", Collections.singletonMap(key,
value));
+ IllegalArgumentException failure =
+ assertThrows(IllegalArgumentException.class, () ->
configuration(config));
+ assertTrue(failure.getMessage().contains(message),
failure.getMessage());
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-s3-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/s3/S3FileConnectDryRunIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-s3-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/s3/S3FileConnectDryRunIT.java
new file mode 100644
index 0000000000..954f64013c
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-s3-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/s3/S3FileConnectDryRunIT.java
@@ -0,0 +1,190 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.connector.file.s3;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.factory.FactoryUtil;
+import org.apache.seatunnel.api.table.factory.SupportSourceDryRunValidation;
+import org.apache.seatunnel.api.table.factory.TableSourceFactory;
+import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
+import org.apache.seatunnel.e2e.common.TestResource;
+import org.apache.seatunnel.e2e.common.TestSuiteBase;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.DockerLoggerFactory;
+
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.client.builder.AwsClientBuilder;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.model.AmazonS3Exception;
+
+import java.io.FileNotFoundException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Exercises the source factory metadata contract against MinIO without
submitting a job. */
+@Timeout(60)
+public class S3FileConnectDryRunIT extends TestSuiteBase implements
TestResource {
+ private static final String IMAGE =
"minio/minio:RELEASE.2024-06-13T22-53-53Z";
+ private static final String BUCKET = "dry-run-events";
+ private static final String ACCESS_KEY = "minioadmin";
+ private static final String SECRET_KEY = "minioadmin";
+ private GenericContainer<?> minio;
+ private AmazonS3 admin;
+ private String endpoint;
+
+ @BeforeAll
+ @Override
+ public void startUp() {
+ minio =
+ new GenericContainer<>(DockerImageName.parse(IMAGE))
+ .withEnv("MINIO_ROOT_USER", ACCESS_KEY)
+ .withEnv("MINIO_ROOT_PASSWORD", SECRET_KEY)
+ .withCommand("server", "/data")
+ .withExposedPorts(9000)
+
.waitingFor(Wait.forHttp("/minio/health/ready").forPort(9000))
+ .withStartupTimeout(Duration.ofMinutes(2))
+ .withLogConsumer(
+ new
Slf4jLogConsumer(DockerLoggerFactory.getLogger(IMAGE)));
+ minio.start();
+ endpoint = "http://"; + minio.getHost() + ":" +
minio.getMappedPort(9000);
+ admin =
+ AmazonS3ClientBuilder.standard()
+ .withCredentials(
+ new AWSStaticCredentialsProvider(
+ new BasicAWSCredentials(ACCESS_KEY,
SECRET_KEY)))
+ .withEndpointConfiguration(
+ new
AwsClientBuilder.EndpointConfiguration(endpoint, "us-east-1"))
+ .withPathStyleAccessEnabled(true)
+ .build();
+ admin.createBucket(BUCKET);
+ admin.putObject(BUCKET, "events/data.json", "{\"event_id\":1}");
+ admin.createBucket("dry-run-empty");
+ }
+
+ @AfterAll
+ @Override
+ public void tearDown() {
+ try {
+ if (admin != null) {
+ admin.shutdown();
+ }
+ } finally {
+ if (minio != null) {
+ minio.stop();
+ }
+ }
+ }
+
+ @Test
+ void shouldValidateObjectAndPrefixWithoutChangingStoredData() throws
Exception {
+ String etag = admin.getObjectMetadata(BUCKET,
"events/data.json").getETag();
+ validate(sourceConfig(endpoint, BUCKET, "/events/data.json"));
+ validate(sourceConfig(endpoint, BUCKET, "/events"));
+ assertEquals(etag, admin.getObjectMetadata(BUCKET,
"events/data.json").getETag());
+ assertEquals("{\"event_id\":1}", admin.getObjectAsString(BUCKET,
"events/data.json"));
+ assertEquals(1, admin.listObjectsV2(BUCKET).getKeyCount());
+ }
+
+ @Test
+ void shouldRejectMissingBatchPrefix() {
+ FileNotFoundException failure =
+ assertThrows(
+ FileNotFoundException.class,
+ () -> validate(sourceConfig(endpoint, BUCKET,
"/future")));
+ assertTrue(failure.getMessage().contains("batch source path"));
+ }
+
+ @Test
+ void shouldAcceptEmptyContinuousPrefix() {
+ Map<String, Object> config = sourceConfig(endpoint, BUCKET, "/future");
+ config.put("discovery_mode", "continuous");
+ assertDoesNotThrow(() -> validate(config));
+ }
+
+ @Test
+ void shouldAcceptAccessibleEmptyBucketRoot() {
+ assertDoesNotThrow(() -> validate(sourceConfig(endpoint,
"dry-run-empty", "/")));
+ }
+
+ @Test
+ void shouldRejectMissingBucketInContinuousMode() {
+ Map<String, Object> config = sourceConfig(endpoint, "dry-run-missing",
"/future");
+ config.put("discovery_mode", "continuous");
+ AmazonS3Exception failure = assertThrows(AmazonS3Exception.class, ()
-> validate(config));
+ assertEquals(404, failure.getStatusCode());
+ assertEquals("NoSuchBucket", failure.getErrorCode());
+ }
+
+ @Test
+ void shouldRejectIncorrectCredentials() {
+ Map<String, Object> config = sourceConfig(endpoint, BUCKET,
"/events/data.json");
+ config.put("secret_key", "incorrect-test-secret");
+ AmazonS3Exception failure = assertThrows(AmazonS3Exception.class, ()
-> validate(config));
+ assertEquals(403, failure.getStatusCode());
+ assertTrue(failure.getMessage().contains("403"));
+ }
+
+ static Map<String, Object> sourceConfig(String endpoint, String bucket,
String path) {
+ Map<String, Object> config = new HashMap<>();
+ config.put("path", path);
+ config.put("file_format_type", "json");
+ config.put("bucket", "s3a://" + bucket);
+ config.put("fs.s3a.endpoint", endpoint);
+ config.put(
+ "fs.s3a.aws.credentials.provider",
+ "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider");
+ config.put("access_key", ACCESS_KEY);
+ config.put("secret_key", SECRET_KEY);
+ config.put("parse_partition_from_path", false);
+ config.put(
+ "hadoop_s3_properties",
+ Collections.singletonMap("fs.s3a.path.style.access", "true"));
+ config.put(
+ "schema",
+ Collections.singletonMap("fields",
Collections.singletonMap("event_id", "bigint")));
+ return config;
+ }
+
+ static void validate(Map<String, Object> config) throws Exception {
+ ClassLoader classLoader = S3FileConnectDryRunIT.class.getClassLoader();
+ TableSourceFactory factory =
+ FactoryUtil.discoverFactory(classLoader,
TableSourceFactory.class, "S3File");
+ assertTrue(factory instanceof SupportSourceDryRunValidation);
+ SupportSourceDryRunValidation validator =
(SupportSourceDryRunValidation) factory;
+ TableSourceFactoryContext context =
+ new TableSourceFactoryContext(ReadonlyConfig.fromMap(config),
classLoader);
+ validator.validateConnectionForDryRun(context,
validator.inferSchemaForDryRun(context));
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-s3-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/s3/S3FileConnectDryRunWireIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-s3-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/s3/S3FileConnectDryRunWireIT.java
new file mode 100644
index 0000000000..2cedd84c2c
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-s3-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/s3/S3FileConnectDryRunWireIT.java
@@ -0,0 +1,249 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.connector.file.s3;
+
+import org.apache.seatunnel.e2e.common.TestSuiteBase;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import com.amazonaws.AmazonClientException;
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.services.s3.model.AmazonS3Exception;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Records actual SDK requests; no SeaTunnel jobs or object-content requests
are used. */
+@Timeout(30)
+public class S3FileConnectDryRunWireIT extends TestSuiteBase {
+ @Test
+ void shouldUseOnlyObjectHeadForExactPath() throws Exception {
+ try (MetadataServer server = new MetadataServer("object")) {
+ S3FileConnectDryRunIT.validate(server.config());
+ assertEquals(Arrays.asList("HEAD /dry-run-events/events"),
server.requests());
+ }
+ }
+
+ @Test
+ void shouldListOnlyOnePrefixEntryWithoutFetchingContentsOrNextPage()
throws Exception {
+ try (MetadataServer server = new MetadataServer("prefix")) {
+ assertDoesNotThrow(
+ () -> S3FileConnectDryRunIT.validate(server.config()),
+ () -> server.requests().toString());
+ List<String> requests = server.requests();
+ assertEquals(2, requests.size());
+ assertEquals("HEAD /dry-run-events/events", requests.get(0));
+ assertTrue(requests.get(1).startsWith("GET /dry-run-events/?"),
requests.toString());
+ assertTrue(requests.get(1).contains("max-keys=1"));
+ assertTrue(requests.get(1).contains("prefix=events%2F"));
+ assertTrue(requests.get(1).contains("delimiter=%2F"));
+ assertFalse(requests.get(1).contains("continuation-token"));
+ }
+ }
+
+ @Test
+ void shouldPropagateForbiddenMetadataWithoutRetrying() throws Exception {
+ try (MetadataServer server = new MetadataServer("forbidden")) {
+ AmazonS3Exception failure =
+ assertThrows(
+ AmazonS3Exception.class,
+ () ->
S3FileConnectDryRunIT.validate(server.config()));
+ assertEquals(403, failure.getStatusCode());
+ assertEquals(1, server.requests().size());
+ }
+ }
+
+ @Test
+ void shouldUseConfiguredSmallerSocketTimeoutAndCloseCredentialsOnFailure()
throws Exception {
+ try (MetadataServer server = new MetadataServer("stalled")) {
+ Map<String, Object> config = server.config();
+ Map<String, String> properties = new HashMap<>();
+ properties.put("fs.s3a.path.style.access", "true");
+ properties.put("fs.s3a.connection.timeout", "200");
+ config.put("hadoop_s3_properties", properties);
+ config.put(
+ "fs.s3a.aws.credentials.provider",
TrackingCredentialsProvider.class.getName());
+ int closedBefore = TrackingCredentialsProvider.CLOSED.get();
+ AmazonClientException failure =
+ assertThrows(
+ AmazonClientException.class,
+ () -> S3FileConnectDryRunIT.validate(config));
+ assertTrue(failure.getMessage().contains("Unable to execute HTTP
request"));
+ assertEquals(1, server.requests().size());
+ assertEquals(closedBefore + 1,
TrackingCredentialsProvider.CLOSED.get());
+ }
+ }
+
+ @Test
+ void shouldCloseCredentialsAfterSuccessfulMetadataRequest() throws
Exception {
+ try (MetadataServer server = new MetadataServer("object")) {
+ Map<String, Object> config = server.config();
+ config.put(
+ "fs.s3a.aws.credentials.provider",
TrackingCredentialsProvider.class.getName());
+ int closedBefore = TrackingCredentialsProvider.CLOSED.get();
+ S3FileConnectDryRunIT.validate(config);
+ assertEquals(closedBefore + 1,
TrackingCredentialsProvider.CLOSED.get());
+ }
+ }
+
+ public static class TrackingCredentialsProvider implements
AWSCredentialsProvider, Closeable {
+ private static final AtomicInteger CLOSED = new AtomicInteger();
+ private static final AtomicInteger CREATED = new AtomicInteger();
+
+ public TrackingCredentialsProvider() {
+ CREATED.incrementAndGet();
+ }
+
+ @Override
+ public AWSCredentials getCredentials() {
+ return new BasicAWSCredentials("minioadmin", "minioadmin");
+ }
+
+ @Override
+ public void refresh() {}
+
+ @Override
+ public void close() {
+ CLOSED.incrementAndGet();
+ }
+ }
+
+ @Test
+ void shouldCloseCredentialsWhenEndpointInitializationFails() {
+ Map<String, Object> config =
+ S3FileConnectDryRunIT.sourceConfig("http://[";,
"dry-run-events", "/events");
+ config.put("fs.s3a.aws.credentials.provider",
TrackingCredentialsProvider.class.getName());
+ int closedBefore = TrackingCredentialsProvider.CLOSED.get();
+ IllegalArgumentException failure =
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> S3FileConnectDryRunIT.validate(config));
+ assertTrue(failure.getMessage().contains("endpoint"));
+ assertEquals(closedBefore + 1,
TrackingCredentialsProvider.CLOSED.get());
+ }
+
+ @Test
+ void shouldRejectInvalidClientSettingsBeforeCreatingCredentials() {
+ Map<String, Object> config =
+ S3FileConnectDryRunIT.sourceConfig("http://localhost";,
"dry-run-events", "/events");
+ config.put("fs.s3a.aws.credentials.provider",
TrackingCredentialsProvider.class.getName());
+ Map<String, String> properties = new HashMap<>();
+ properties.put("fs.s3a.connection.maximum", "0");
+ config.put("hadoop_s3_properties", properties);
+ int createdBefore = TrackingCredentialsProvider.CREATED.get();
+ IllegalArgumentException failure =
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> S3FileConnectDryRunIT.validate(config));
+ assertTrue(failure.getMessage().contains("fs.s3a.connection.maximum"));
+ assertEquals(createdBefore, TrackingCredentialsProvider.CREATED.get());
+ }
+
+ private static final class MetadataServer implements AutoCloseable {
+ private final HttpServer server;
+ private final ExecutorService executor =
Executors.newCachedThreadPool();
+ private final ConcurrentLinkedQueue<String> requests = new
ConcurrentLinkedQueue<>();
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ private MetadataServer(String response) throws IOException {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0),
0);
+ server.setExecutor(executor);
+ server.createContext("/", exchange -> reply(exchange, response));
+ server.start();
+ }
+
+ private void reply(HttpExchange exchange, String response) throws
IOException {
+ requests.add(exchange.getRequestMethod() + " " +
exchange.getRequestURI());
+ try {
+ exchange.getResponseHeaders().add("Connection", "close");
+ if ("stalled".equals(response)) {
+ try {
+ release.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return;
+ }
+ if ("HEAD".equals(exchange.getRequestMethod())) {
+ int status =
+ "object".equals(response)
+ ? 200
+ : "forbidden".equals(response) ? 403 : 404;
+ exchange.getResponseHeaders().add("ETag",
"\"event-etag\"");
+ exchange.getResponseHeaders()
+ .add("Last-Modified", "Tue, 08 Sep 2026 00:00:00
GMT");
+ exchange.sendResponseHeaders(status, -1);
+ return;
+ }
+ String xml =
+ "<ListBucketResult
xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\";>"
+ +
"<Name>dry-run-events</Name><Prefix>events/</Prefix><MaxKeys>1</MaxKeys>"
+ +
"<IsTruncated>true</IsTruncated><NextContinuationToken>remaining-events</NextContinuationToken>"
+ +
"<Contents><Key>events/data.json</Key><LastModified>2026-09-08T00:00:00.000Z</LastModified>"
+ +
"<ETag>event-etag</ETag><Size>14</Size><StorageClass>STANDARD</StorageClass></Contents></ListBucketResult>";
+ byte[] body = xml.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type",
"application/xml");
+ exchange.sendResponseHeaders(200, body.length);
+ exchange.getResponseBody().write(body);
+ } finally {
+ exchange.close();
+ }
+ }
+
+ private Map<String, Object> config() {
+ return S3FileConnectDryRunIT.sourceConfig(
+ "http://127.0.0.1:"; + server.getAddress().getPort(),
+ "dry-run-events",
+ "/events");
+ }
+
+ private List<String> requests() {
+ return new ArrayList<>(requests);
+ }
+
+ @Override
+ public void close() {
+ release.countDown();
+ server.stop(0);
+ executor.shutdownNow();
+ }
+ }
+}
