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 4492252928 [Fix][Connector-V2] Declare missing source options in file
source factories (#12817)
4492252928 is described below
commit 449225292833a7f8c5e0b0a9169a28b855245c86
Author: chovygo <[email protected]>
AuthorDate: Sun Oct 18 06:25:03 2026 +0000
[Fix][Connector-V2] Declare missing source options in file source factories
(#12817)
Co-authored-by: zhaowei <[email protected]>
---
docs/en/connectors/source/GcsFile.md | 6 +++
docs/en/connectors/source/OssFile.md | 1 +
docs/en/connectors/source/OssJindoFile.md | 2 +
docs/en/connectors/source/SmbFile.md | 7 +++
docs/zh/connectors/source/GcsFile.md | 6 +++
docs/zh/connectors/source/OssFile.md | 1 +
docs/zh/connectors/source/OssJindoFile.md | 2 +
docs/zh/connectors/source/SmbFile.md | 7 +++
.../file/bos/source/BosFileSourceFactory.java | 9 ++++
.../seatunnel/file/bos/BosFileFactoryTest.java | 31 ++++++++++++
.../file/cos/source/CosFileSourceFactory.java | 9 ++++
.../seatunnel/file/cos/CosFileFactoryTest.java | 31 ++++++++++++
.../file/ftp/source/FtpFileSourceFactory.java | 9 ++++
.../seatunnel/file/ftp/FtpFileFactoryTest.java | 29 ++++++++++++
.../file/gcs/source/GcsFileSourceFactory.java | 9 ++++
.../file/gcs/GcsFileSourceFactoryTest.java | 28 +++++++++++
.../file/hdfs/source/HdfsFileSourceFactory.java | 9 ++++
.../seatunnel/file/hdfs/HdfsFileFactoryTest.java | 29 ++++++++++++
.../oss/jindo/source/OssFileSourceFactory.java | 10 ++++
.../connectors/test/OssJindoFactoryTest.java | 52 ++++++++++++++++++++
.../file/local/source/LocalFileSourceFactory.java | 9 ++++
.../seatunnel/file/local/LocalFileFactoryTest.java | 36 ++++++++++++++
.../file/obs/source/ObsFileSourceFactory.java | 11 +++++
.../seatunnel/file/obs/ObsFileFactoryTest.java | 55 +++++++++++++++++++++-
.../file/oss/source/OssFileSourceFactory.java | 9 ++++
.../seatunnel/file/oss/OssFileFactoryTest.java | 31 ++++++++++++
.../file/s3/source/S3FileSourceFactory.java | 9 ++++
.../seatunnel/file/s3/S3FileFactoryTest.java | 29 ++++++++++++
.../file/sftp/source/SftpFileSourceFactory.java | 11 +++++
.../seatunnel/file/sftp/SftpFileFactoryTest.java | 44 +++++++++++++++++
.../file/smb/source/SmbFileSourceFactory.java | 11 +++++
.../seatunnel/file/smb/SmbFileFactoryTest.java | 44 +++++++++++++++++
32 files changed, 585 insertions(+), 1 deletion(-)
diff --git a/docs/en/connectors/source/GcsFile.md
b/docs/en/connectors/source/GcsFile.md
index 260b3147ed..b95c30d397 100644
--- a/docs/en/connectors/source/GcsFile.md
+++ b/docs/en/connectors/source/GcsFile.md
@@ -74,6 +74,8 @@ The explicit `service_account_key_file` option takes
precedence over the corresp
| enable_file_split | boolean | no | `false` | Enables logical splitting for
uncompressed text, CSV, JSON, and Parquet files. |
| file_split_size | long | conditional | `134217728` | Split size in bytes
when `enable_file_split=true`. |
| null_format | string | no | - | Text representation of null values. |
+| binary_chunk_size | int | no | 1024 | Only used when file_format_type is
binary. The chunk size (in bytes) for reading binary files. Larger values may
improve performance for large files but use more memory. |
+| binary_complete_file_mode | boolean | no | false | Only used when
file_format_type is binary. Whether to read the complete file as a single chunk
instead of splitting into chunks. |
| quote_char | string | no | `"` | CSV quote character. |
| escape_char | string | no | - | CSV escape character. |
| sheet_name | string | no | - | Excel worksheet to read. |
@@ -81,6 +83,7 @@ The explicit `service_account_key_file` option takes
precedence over the corresp
| poi_excel_max_file_size | long | no | `52428800` | Maximum Excel file size
in bytes for the POI engine. |
| xml_row_tag | string | conditional | - | XML element representing one row. |
| xml_use_attr_format | boolean | conditional | - | Whether XML values are
read from attributes. |
+| csv_use_header_line | boolean | no | false | Whether to use the header line
to parse the file, only used when the file_format is `csv` and the file
contains the header line that match RFC 4180. |
| discovery_mode | string | no | `once` | `once` or `continuous`. Continuous
mode currently requires update sync and binary format. |
| scan_interval | string | no | `10S` | Poll interval for continuous
discovery. |
| start_mode | string | no | `earliest` | `earliest` processes existing files;
`latest` starts with later changes. |
@@ -95,6 +98,9 @@ The explicit `service_account_key_file` option takes
precedence over the corresp
| backup_path | string | conditional | - | Required for
`post_sync_action=backup`. Must not overlap the source path. |
| retention_max_age | string | no | - | Maximum age for SeaTunnel backup
objects. |
| retention_check_interval | string | no | `1H` | Backup retention scan
interval. |
+| file_filter_modified_start | string | no | - | File modification time
filter. The connector will filter files based on the last modification start
time (inclusive). Default format: `yyyy-MM-dd HH:mm:ss`. |
+| file_filter_modified_end | string | no | - | File modification time filter.
The connector will filter files based on the last modification end time
(exclusive). Default format: `yyyy-MM-dd HH:mm:ss`. |
+| sort_files_by_modification_time | boolean | no | false | Sort files by
modification time in descending order. Enable this when reading evolving
schemas to ensure schema inference uses the latest file. |
| common-options | | no | - | See [Source Common
Options](../common-options/source-common-options.md). |
## Example
diff --git a/docs/en/connectors/source/OssFile.md
b/docs/en/connectors/source/OssFile.md
index 8d937d0d09..2e3b8357cc 100644
--- a/docs/en/connectors/source/OssFile.md
+++ b/docs/en/connectors/source/OssFile.md
@@ -210,6 +210,7 @@ If you assign file type to `parquet` `orc`, schema option
not required, connecto
| xml_use_attr_format | boolean | no | - |
Specifies whether to process data using the tag attribute format, only used
when file_format is xml.
|
| csv_use_header_line | boolean | no | false |
Whether to use the header line to parse the file, only used when the
file_format is `csv` and the file contains the header line that match RFC 4180
|
| compress_codec | string | no | none |
Which compress codec the files used.
|
+| archive_compress_codec | string | no | none |
Which archive compress codec the files used. Supported: `none` `zip` `tar`
`tar.gz` `gz`.
|
| encoding | string | no | UTF-8 | File
encoding, only used when `file_format_type` is `json`, `text`, `csv`, or `xml`.
|
| null_format | string | no | - | Only
used when file_format_type is text. null_format to define which strings can be
represented as null. e.g: `\N`
|
| binary_chunk_size | int | no | 1024 | Only
used when file_format_type is binary. The chunk size (in bytes) for reading
binary files. Default is 1024 bytes. Larger values may improve performance for
large files but use more memory.
|
diff --git a/docs/en/connectors/source/OssJindoFile.md
b/docs/en/connectors/source/OssJindoFile.md
index ffd62381c1..af07373807 100644
--- a/docs/en/connectors/source/OssJindoFile.md
+++ b/docs/en/connectors/source/OssJindoFile.md
@@ -87,6 +87,8 @@ It only supports hadoop version **2.9.X+**.
| archive_compress_codec | string | no | none
|
| encoding | string | no | UTF-8
|
| null_format | string | no | -
|
+| binary_chunk_size | int | no | 1024
|
+| binary_complete_file_mode | boolean | no | false
|
| common-options | | no | -
|
| file_filter_modified_start | string | no | -
|
| file_filter_modified_end | string | no | -
|
diff --git a/docs/en/connectors/source/SmbFile.md
b/docs/en/connectors/source/SmbFile.md
index 94ae3bdeac..f24e4cddea 100644
--- a/docs/en/connectors/source/SmbFile.md
+++ b/docs/en/connectors/source/SmbFile.md
@@ -99,9 +99,13 @@ The File does not have a specific type list, and we can
indicate which SeaTunnel
| sheet_name | String | No | -
| Reader the sheet of the workbook, Only used when file_format is excel.
|
| xml_row_tag | String | No | -
| Specifies the tag name of the data rows within the XML file, only used
when file_format is xml. |
| xml_use_attr_format | Boolean | No | -
| Specifies whether to process data using the tag attribute format, only
used when file_format is xml. |
+| csv_use_header_line | boolean | no | false | Whether to use the header line
to parse the file, only used when the file_format is `csv` and the file
contains the header line that match RFC 4180. |
| compress_codec | String | No | None
| The compress codec of files
|
+| archive_compress_codec | String | No | none
| Which archive compress codec the files used. Supported: `none` `zip` `tar`
`tar.gz` `gz`. |
| encoding | String | No | UTF-8
| The encoding of the file to read
|
| null_format | String | No | -
| Only used when file_format_type is text. Define which strings can be
represented as null, e.g. `\N`. |
+| binary_chunk_size | int | no | 1024 | Only used when file_format_type is
binary. The chunk size (in bytes) for reading binary files. Larger values may
improve performance for large files but use more memory. |
+| binary_complete_file_mode | boolean | no | false | Only used when
file_format_type is binary. Whether to read the complete file as a single chunk
instead of splitting into chunks. |
| filename_extension | String | No | -
| Filter filename extension, which used for filtering files with specific
extension. Example: `csv` `.txt` `json` `.xml`. |
| excel_engine | String | No | POI
| Only used when file_format is excel. Supported engines are `POI` and
`EasyExcel`. |
| poi_excel_max_file_size | Long | No | 52428800
| Only used when file_format is excel and excel_engine is POI. The maximum
Excel file size in bytes (default 50 MB). |
@@ -123,6 +127,9 @@ The File does not have a specific type list, and we can
indicate which SeaTunnel
| retention_max_age | String | No | -
| Optional retention age for backup files, only valid when
`post_sync_action=backup`.
|
| retention_check_interval | String | No | 1H
| Retention scan interval, only effective when `post_sync_action=backup` and
`retention_max_age` is configured. |
| recursive_file_scan | Boolean | No | true
| Whether to scan subdirectories recursively. If `false`, subdirectories
will be ignored. |
+| file_filter_modified_start | string | no | - | File modification time
filter. The connector will filter files based on the last modification start
time (inclusive). Default format: `yyyy-MM-dd HH:mm:ss`. |
+| file_filter_modified_end | string | no | - | File modification time filter.
The connector will filter files based on the last modification end time
(exclusive). Default format: `yyyy-MM-dd HH:mm:ss`. |
+| sort_files_by_modification_time | boolean | no | false | Sort files by
modification time in descending order. Enable this when reading evolving
schemas to ensure schema inference uses the latest file. |
| common-options | | No | -
| Source plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details.
|
## How to Create a SMB Data Synchronization Job
diff --git a/docs/zh/connectors/source/GcsFile.md
b/docs/zh/connectors/source/GcsFile.md
index 0aa7f4a74e..e60bc57e53 100644
--- a/docs/zh/connectors/source/GcsFile.md
+++ b/docs/zh/connectors/source/GcsFile.md
@@ -64,6 +64,8 @@ import ChangeLog from '../changelog/connector-file-gcs.md';
| enable_file_split | boolean | 否 | `false` | 为未压缩的 text、CSV、JSON 和 Parquet
文件启用逻辑切分。 |
| file_split_size | long | 条件必填 | `134217728` | `enable_file_split=true`
时的切分大小，单位为字节。 |
| null_format | string | 否 | - | 表示空值的文本。 |
+| binary_chunk_size | int | 否 | 1024 | 仅在 file_format_type 为 binary
时使用。读取二进制文件的块大小（以字节为单位）。较大的值可能会提高大文件的性能，但会使用更多内存。 |
+| binary_complete_file_mode | boolean | 否 | false | 仅在 file_format_type 为
binary 时使用。是否将完整文件作为单个块读取，而不是分割成块。 |
| quote_char | string | 否 | `"` | CSV 引号字符。 |
| escape_char | string | 否 | - | CSV 转义字符。 |
| sheet_name | string | 否 | - | 要读取的 Excel 工作表。 |
@@ -71,6 +73,7 @@ import ChangeLog from '../changelog/connector-file-gcs.md';
| poi_excel_max_file_size | long | 否 | `52428800` | POI 引擎可读取的最大 Excel 文件字节数。 |
| xml_row_tag | string | 条件必填 | - | 表示一行数据的 XML 元素。 |
| xml_use_attr_format | boolean | 条件必填 | - | 是否从 XML 属性读取值。 |
+| csv_use_header_line | boolean | 否 | false | 是否使用标题行来解析文件，仅在 file_format 为
`csv` 且文件包含符合 RFC 4180 的标题行时使用。 |
| discovery_mode | string | 否 | `once` | `once` 或 `continuous`。连续发现当前要求 update
同步和 binary 格式。 |
| scan_interval | string | 否 | `10S` | 连续发现的轮询间隔。 |
| start_mode | string | 否 | `earliest` | `earliest` 处理已有文件，`latest` 从后续变更开始。 |
@@ -85,6 +88,9 @@ import ChangeLog from '../changelog/connector-file-gcs.md';
| backup_path | string | 条件必填 | - | `post_sync_action=backup` 时必填，且不能与源路径重叠。 |
| retention_max_age | string | 否 | - | SeaTunnel 备份对象的最大保留时间。 |
| retention_check_interval | string | 否 | `1H` | 备份保留扫描间隔。 |
+| file_filter_modified_start | string | 否 | - |
按文件最后修改时间筛选文件的起始时间（包含该时间）。默认格式：`yyyy-MM-dd HH:mm:ss`。 |
+| file_filter_modified_end | string | 否 | - |
按文件最后修改时间筛选文件的结束时间（不包含该时间）。默认格式：`yyyy-MM-dd HH:mm:ss`。 |
+| sort_files_by_modification_time | boolean | 否 | false |
是否按修改时间降序排序文件。启用此选项后，在读取不断演化的 schema 时可确保 schema 推断使用最新的文件。 |
| common-options | | 否 | - | 参见 [Source
通用配置](../common-options/source-common-options.md)。 |
## 示例
diff --git a/docs/zh/connectors/source/OssFile.md
b/docs/zh/connectors/source/OssFile.md
index f7231c21bb..a3b5915e25 100644
--- a/docs/zh/connectors/source/OssFile.md
+++ b/docs/zh/connectors/source/OssFile.md
@@ -210,6 +210,7 @@ schema {
| xml_row_tag | string | 否 | - |
指定XML文件中数据行的标签名称，仅在file_format为xml时使用。
|
| xml_use_attr_format | boolean | 否 | - |
指定是否使用标签属性格式处理数据，仅在file_format为xml时使用。
|
| compress_codec | string | 否 | none |
文件使用的压缩编解码器。
|
+| archive_compress_codec | string | 否 | none |
文件使用的归档压缩编解码器。支持：`none` `zip` `tar` `tar.gz` `gz`。
|
| encoding | string | 否 | UTF-8 | 文件编码，仅在
`file_format_type` 为 `json`、`text`、`csv` 或 `xml` 时使用。 |
| null_format | string | 否 | - |
仅在file_format_type为text时使用。null_format用于定义哪些字符串可以表示为null。例如：`\N`
|
| binary_chunk_size | int | 否 | 1024 |
仅在file_format_type为binary时使用。读取二进制文件的块大小（以字节为单位）。默认为1024字节。较大的值可能会提高大文件的性能，但会使用更多内存。
|
diff --git a/docs/zh/connectors/source/OssJindoFile.md
b/docs/zh/connectors/source/OssJindoFile.md
index b94af55993..a872c98592 100644
--- a/docs/zh/connectors/source/OssJindoFile.md
+++ b/docs/zh/connectors/source/OssJindoFile.md
@@ -87,6 +87,8 @@ import ChangeLog from
'../changelog/connector-file-oss-jindo.md';
| archive_compress_codec | string | 否 | none |
读取归档文件时使用的归档压缩格式。 |
| encoding | string | 否 | UTF-8 | 读取
`json`、`text`、`csv` 或 `xml` 文件时使用的字符编码。 |
| null_format | string | 否 | - | 仅用于
`text` 格式，指定应解析为 `null` 的字符串。 |
+| binary_chunk_size | int | 否 | 1024 | 仅在
file_format_type 为 binary 时使用。读取二进制文件的块大小（以字节为单位）。默认为 1024
字节。较大的值可能会提高大文件的性能，但会使用更多内存。 |
+| binary_complete_file_mode | boolean | 否 | false | 仅在
file_format_type 为 binary 时使用。是否将完整文件作为单个块读取，而不是分割成块。 |
| common-options | | 否 | - |
Source 插件通用参数，详见 [Source Common
Options](../common-options/source-common-options.md)。 |
| file_filter_modified_start | string | 否 | - |
按文件最后修改时间筛选文件的起始时间（包含该时间）。 |
| file_filter_modified_end | string | 否 | - |
按文件最后修改时间筛选文件的结束时间（不包含该时间）。 |
diff --git a/docs/zh/connectors/source/SmbFile.md
b/docs/zh/connectors/source/SmbFile.md
index 02f45542dd..6c42be4837 100644
--- a/docs/zh/connectors/source/SmbFile.md
+++ b/docs/zh/connectors/source/SmbFile.md
@@ -99,9 +99,13 @@ import ChangeLog from '../changelog/connector-file-smb.md';
| sheet_name | String | 否 | - |
读取工作簿中的 sheet，仅在 file_format 为 excel 时使用 |
| xml_row_tag | String | 否 | - |
指定 XML 文件中数据行的标签名，仅在 file_format 为 xml 时使用 |
| xml_use_attr_format | Boolean | 否 | - |
指定是否使用标签属性格式处理数据，仅在 file_format 为 xml 时使用 |
+| csv_use_header_line | boolean | 否 | false | 是否使用标题行来解析文件，仅在 file_format 为
`csv` 且文件包含符合 RFC 4180 的标题行时使用 |
| compress_codec | String | 否 | None |
文件的压缩编解码器 |
+| archive_compress_codec | String | 否 | none |
文件使用的归档压缩编解码器。支持：`none` `zip` `tar` `tar.gz` `gz`。 |
| encoding | String | 否 | UTF-8 |
读取文件时使用的编码
|
| null_format | String | 否 | - |
仅在 file_format_type 为 text 时使用。定义哪些字符串可以表示为 null，例如 `\N` |
+| binary_chunk_size | int | 否 | 1024 | 仅在 file_format_type 为 binary
时使用。读取二进制文件的块大小（以字节为单位）。较大的值可能会提高大文件的性能，但会使用更多内存。 |
+| binary_complete_file_mode | boolean | 否 | false | 仅在 file_format_type 为
binary 时使用。是否将完整文件作为单个块读取，而不是分割成块。 |
| filename_extension | String | 否 | - |
文件扩展名过滤，用于过滤特定扩展名的文件。例如：`csv` `.txt` `json` `.xml` |
| excel_engine | String | 否 | POI |
仅在 file_format 为 excel 时使用。支持的引擎为 `POI` 和 `EasyExcel` |
| poi_excel_max_file_size | Long | 否 | 52428800 |
仅在 file_format 为 excel 且 excel_engine 为 POI 时使用。POI 引擎可读取的最大 Excel 文件大小（默认 50
MB） |
@@ -123,6 +127,9 @@ import ChangeLog from '../changelog/connector-file-smb.md';
| retention_max_age | String | 否 | - |
备份文件的可选保留期限，仅在 `post_sync_action=backup` 时有效 |
| retention_check_interval | String | 否 | 1H |
保留扫描间隔，仅在配置了 `post_sync_action=backup` 和 `retention_max_age` 时生效 |
| recursive_file_scan | Boolean | 否 | true |
是否递归扫描子目录。如果为 `false`，将忽略子目录 |
+| file_filter_modified_start | string | 否 | - |
按文件最后修改时间筛选文件的起始时间（包含该时间）。默认格式：`yyyy-MM-dd HH:mm:ss`。 |
+| file_filter_modified_end | string | 否 | - |
按文件最后修改时间筛选文件的结束时间（不包含该时间）。默认格式：`yyyy-MM-dd HH:mm:ss`。 |
+| sort_files_by_modification_time | boolean | 否 | false |
是否按修改时间降序排序文件。启用此选项后，在读取不断演化的 schema 时可确保 schema 推断使用最新的文件。 |
| common-options | | 否 | - |
源插件通用参数，请参阅 [Source Common Options](../common-options/source-common-options.md)
|
## 如何创建 SMB 数据同步作业
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-bos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/bos/source/BosFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-bos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/bos/source/BosFileSourceFactory.java
index 9da0434b51..d3e90a1719 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-bos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/bos/source/BosFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-bos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/bos/source/BosFileSourceFactory.java
@@ -106,6 +106,15 @@ public class BosFileSourceFactory implements
TableSourceFactory {
.optional(FileBaseSourceOptions.QUOTE_CHAR)
.optional(FileBaseSourceOptions.ESCAPE_CHAR)
.optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN)
+ .optional(
+ FileBaseSourceOptions.READ_PARTITIONS,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END,
+ FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)
+ .optional(
+ FileBaseSourceOptions.CSV_USE_HEADER_LINE,
+ FileBaseSourceOptions.BINARY_CHUNK_SIZE,
+ FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-bos/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/bos/BosFileFactoryTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-bos/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/bos/BosFileFactoryTest.java
index 9071d46c97..6123a990b4 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-bos/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/bos/BosFileFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-bos/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/bos/BosFileFactoryTest.java
@@ -17,8 +17,10 @@
package org.apache.seatunnel.connectors.seatunnel.file.bos;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
import
org.apache.seatunnel.connectors.seatunnel.file.bos.sink.BosFileSinkFactory;
import
org.apache.seatunnel.connectors.seatunnel.file.bos.source.BosFileSourceFactory;
+import
org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -30,4 +32,33 @@ public class BosFileFactoryTest {
Assertions.assertNotNull((new BosFileSourceFactory()).optionRule());
Assertions.assertNotNull((new BosFileSinkFactory()).optionRule());
}
+
+ @Test
+ void consumedOptionsAreDeclared() {
+ OptionRule optionRule = (new BosFileSourceFactory()).optionRule();
+ Assertions.assertTrue(
+
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.READ_PARTITIONS));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+ .contains(FileBaseSourceOptions.CSV_USE_HEADER_LINE));
+ Assertions.assertTrue(
+
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.BINARY_CHUNK_SIZE));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_START));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_END));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME));
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-cos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-cos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java
index 26ca989daa..fe6422e75c 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-cos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-cos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java
@@ -106,6 +106,15 @@ public class CosFileSourceFactory implements
TableSourceFactory {
.optional(FileBaseSourceOptions.QUOTE_CHAR)
.optional(FileBaseSourceOptions.ESCAPE_CHAR)
.optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN)
+ .optional(
+ FileBaseSourceOptions.READ_PARTITIONS,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END,
+ FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)
+ .optional(
+ FileBaseSourceOptions.CSV_USE_HEADER_LINE,
+ FileBaseSourceOptions.BINARY_CHUNK_SIZE,
+ FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-cos/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/cos/CosFileFactoryTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-cos/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/cos/CosFileFactoryTest.java
index 6691f5b1f2..2ebf19a486 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-cos/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/cos/CosFileFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-cos/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/cos/CosFileFactoryTest.java
@@ -17,6 +17,8 @@
package org.apache.seatunnel.connectors.seatunnel.file.cos;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import
org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.file.cos.sink.CosFileSinkFactory;
import
org.apache.seatunnel.connectors.seatunnel.file.cos.source.CosFileSourceFactory;
@@ -30,4 +32,33 @@ public class CosFileFactoryTest {
Assertions.assertNotNull((new CosFileSourceFactory()).optionRule());
Assertions.assertNotNull((new CosFileSinkFactory()).optionRule());
}
+
+ @Test
+ void consumedOptionsAreDeclared() {
+ OptionRule optionRule = (new CosFileSourceFactory()).optionRule();
+ Assertions.assertTrue(
+
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.READ_PARTITIONS));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+ .contains(FileBaseSourceOptions.CSV_USE_HEADER_LINE));
+ Assertions.assertTrue(
+
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.BINARY_CHUNK_SIZE));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_START));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_END));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME));
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java
index 269abc69a8..e965ffca14 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java
@@ -143,6 +143,15 @@ public class FtpFileSourceFactory implements
TableSourceFactory {
FilePostSyncAction.BACKUP,
FileBaseSourceOptions.BACKUP_PATH)
.optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN)
+ .optional(
+ FileBaseSourceOptions.READ_PARTITIONS,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END,
+ FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)
+ .optional(
+ FileBaseSourceOptions.CSV_USE_HEADER_LINE,
+ FileBaseSourceOptions.BINARY_CHUNK_SIZE,
+ FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/FtpFileFactoryTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/FtpFileFactoryTest.java
index 82c3229843..9c14684786 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/FtpFileFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/FtpFileFactoryTest.java
@@ -108,6 +108,35 @@ class FtpFileFactoryTest {
Assertions.assertDoesNotThrow(() -> validate(backupConfig,
optionRule));
}
+ @Test
+ void consumedOptionsAreDeclared() {
+ OptionRule optionRule = (new FtpFileSourceFactory()).optionRule();
+ Assertions.assertTrue(
+
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.READ_PARTITIONS));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+ .contains(FileBaseSourceOptions.CSV_USE_HEADER_LINE));
+ Assertions.assertTrue(
+
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.BINARY_CHUNK_SIZE));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_START));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_END));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME));
+ }
+
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-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/GcsFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/GcsFileSourceFactory.java
index 9f50be47ac..945cb2c03f 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/GcsFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/GcsFileSourceFactory.java
@@ -154,6 +154,15 @@ public class GcsFileSourceFactory implements
TableSourceFactory {
FilePostSyncAction.BACKUP,
FileBaseSourceOptions.BACKUP_PATH)
.optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN)
+ .optional(
+ FileBaseSourceOptions.READ_PARTITIONS,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END,
+ FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)
+ .optional(
+ FileBaseSourceOptions.CSV_USE_HEADER_LINE,
+ FileBaseSourceOptions.BINARY_CHUNK_SIZE,
+ FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/GcsFileSourceFactoryTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/GcsFileSourceFactoryTest.java
index 677b3f0ebd..843066c922 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/GcsFileSourceFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/GcsFileSourceFactoryTest.java
@@ -73,6 +73,34 @@ class GcsFileSourceFactoryTest {
assertTrue(optionRuleContains(optionRule,
FileBaseSourceOptions.FILE_SPLIT_SIZE));
}
+ @Test
+ void consumedOptionsAreDeclared() {
+ OptionRule optionRule = new GcsFileSourceFactory().optionRule();
+
assertTrue(optionRule.getOptionalOptions().contains(FileBaseSourceOptions.READ_PARTITIONS));
+ assertTrue(
+ optionRule
+ .getOptionalOptions()
+ .contains(FileBaseSourceOptions.CSV_USE_HEADER_LINE));
+ assertTrue(
+
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.BINARY_CHUNK_SIZE));
+ assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE));
+ assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_START));
+ assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_END));
+ assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME));
+ }
+
private static Map<String, Object> sourceConfig() {
Map<String, Object> config = new HashMap<>();
config.put(GcsFileSourceOptions.FILE_PATH.key(), "/source");
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java
index e3bc352336..b90a7e8aeb 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java
@@ -155,6 +155,15 @@ public class HdfsFileSourceFactory implements
TableSourceFactory {
.optional(HdfsFileSourceOptions.ESCAPE_CHAR)
.optional(ConnectorCommonOptions.METALAKE_TYPE)
.optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN)
+ .optional(
+ FileBaseSourceOptions.READ_PARTITIONS,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END,
+ FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)
+ .optional(
+ FileBaseSourceOptions.CSV_USE_HEADER_LINE,
+ FileBaseSourceOptions.BINARY_CHUNK_SIZE,
+ FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)
.build();
}
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 f52d8abe22..13098145a8 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
@@ -125,6 +125,35 @@ class HdfsFileFactoryTest {
Assertions.assertDoesNotThrow(() -> validate(config, optionRule));
}
+ @Test
+ void consumedOptionsAreDeclared() {
+ OptionRule optionRule = (new HdfsFileSourceFactory()).optionRule();
+ Assertions.assertTrue(
+
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.READ_PARTITIONS));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+ .contains(FileBaseSourceOptions.CSV_USE_HEADER_LINE));
+ Assertions.assertTrue(
+
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.BINARY_CHUNK_SIZE));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_START));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_END));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME));
+ }
+
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-jindo-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/jindo/source/OssFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/jindo/source/OssFileSourceFactory.java
index c833d2b041..f94430a48f 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/jindo/source/OssFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/jindo/source/OssFileSourceFactory.java
@@ -87,6 +87,7 @@ public class OssFileSourceFactory implements
TableSourceFactory {
.optional(FileBaseSourceOptions.TIME_FORMAT_LEGACY)
.optional(FileBaseSourceOptions.FILE_FILTER_PATTERN)
.optional(FileBaseSourceOptions.COMPRESS_CODEC)
+ .optional(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC)
.optional(FileBaseSourceOptions.NULL_FORMAT)
.optional(FileBaseSourceOptions.FILENAME_EXTENSION)
.optional(FileBaseSourceOptions.READ_COLUMNS)
@@ -105,6 +106,15 @@ public class OssFileSourceFactory implements
TableSourceFactory {
.optional(FileBaseSourceOptions.QUOTE_CHAR)
.optional(FileBaseSourceOptions.ESCAPE_CHAR)
.optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN)
+ .optional(
+ FileBaseSourceOptions.READ_PARTITIONS,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END,
+ FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)
+ .optional(
+ FileBaseSourceOptions.CSV_USE_HEADER_LINE,
+ FileBaseSourceOptions.BINARY_CHUNK_SIZE,
+ FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/test/java/org/apache/seatunnel/connectors/test/OssJindoFactoryTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/test/java/org/apache/seatunnel/connectors/test/OssJindoFactoryTest.java
index c014f1d492..f5984d9bec 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/test/java/org/apache/seatunnel/connectors/test/OssJindoFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/test/java/org/apache/seatunnel/connectors/test/OssJindoFactoryTest.java
@@ -17,16 +17,68 @@
package org.apache.seatunnel.connectors.test;
+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.connectors.seatunnel.file.config.FileBaseOptions;
+import
org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.file.oss.jindo.sink.OssFileSinkFactory;
import
org.apache.seatunnel.connectors.seatunnel.file.oss.jindo.source.OssFileSourceFactory;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
public class OssJindoFactoryTest {
@Test
public void testOptionRule() {
Assertions.assertNotNull((new OssFileSourceFactory()).optionRule());
Assertions.assertNotNull((new OssFileSinkFactory()).optionRule());
}
+
+ @Test
+ void consumedOptionsAcceptedByValidateUnknownKeys() {
+ // These options are consumed by AbstractReadStrategy /
CsvReadStrategy /
+ // BinaryReadStrategy at runtime. They MUST be declared in the
factory's
+ // optionRule() so that ConfigValidator.validateUnknownKeys (called by
+ // --check / --dry-run at SeaTunnelConfValidateCommand:325) does not
+ // reject them as "unknown option keys".
+ Map<String, Object> consumedOptions = new LinkedHashMap<>();
+ consumedOptions.put(
+ FileBaseSourceOptions.READ_PARTITIONS.key(),
+ Collections.singletonList("year=2024"));
+ consumedOptions.put(FileBaseSourceOptions.CSV_USE_HEADER_LINE.key(),
true);
+ consumedOptions.put(FileBaseSourceOptions.BINARY_CHUNK_SIZE.key(),
2048);
+
consumedOptions.put(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE.key(),
true);
+ consumedOptions.put(
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START.key(),
"2024-01-01 00:00:00");
+ consumedOptions.put(
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END.key(),
"2024-12-31 23:59:59");
+
consumedOptions.put(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME.key(), true);
+ consumedOptions.put(FileBaseSourceOptions.COMPRESS_CODEC.key(),
"gzip");
+
consumedOptions.put(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC.key(), "zip");
+
+ OptionRule optionRule = (new OssFileSourceFactory()).optionRule();
+
+ for (Map.Entry<String, Object> entry : consumedOptions.entrySet()) {
+ Map<String, Object> config = sourceConfig();
+ config.put(entry.getKey(), entry.getValue());
+
+ Assertions.assertDoesNotThrow(
+ () ->
+ ConfigValidator.validateUnknownKeys(
+ ReadonlyConfig.fromMap(config),
optionRule, "OssFile"),
+ "validateUnknownKeys should accept declared option: " +
entry.getKey());
+ }
+ }
+
+ private static Map<String, Object> sourceConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put(FileBaseOptions.FILE_PATH.key(), "/source");
+ return config;
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java
index de77146f4a..b6ef0900ba 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java
@@ -150,6 +150,15 @@ public class LocalFileSourceFactory implements
TableSourceFactory {
FilePostSyncAction.BACKUP,
FileBaseSourceOptions.BACKUP_PATH)
.optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN)
+ .optional(
+ FileBaseSourceOptions.READ_PARTITIONS,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END,
+ FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)
+ .optional(
+ FileBaseSourceOptions.CSV_USE_HEADER_LINE,
+ FileBaseSourceOptions.BINARY_CHUNK_SIZE,
+ FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileFactoryTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileFactoryTest.java
index c4c40323f9..20f32463c8 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileFactoryTest.java
@@ -30,7 +30,9 @@ import
org.apache.seatunnel.connectors.seatunnel.file.local.source.LocalFileSour
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.Map;
class LocalFileFactoryTest {
@@ -118,6 +120,40 @@ class LocalFileFactoryTest {
Assertions.assertDoesNotThrow(() -> validate(backupConfig,
optionRule));
}
+ @Test
+ void consumedOptionsAcceptedByValidateUnknownKeys() {
+ // These 7 options are consumed by AbstractReadStrategy /
CsvReadStrategy /
+ // BinaryReadStrategy at runtime. They MUST be declared in the
factory's
+ // optionRule() so that ConfigValidator.validateUnknownKeys (called by
+ // --check / --dry-run at SeaTunnelConfValidateCommand:325) does not
+ // reject them as "unknown option keys".
+ Map<String, Object> consumedOptions = new LinkedHashMap<>();
+ consumedOptions.put(
+ FileBaseSourceOptions.READ_PARTITIONS.key(),
+ Collections.singletonList("year=2024"));
+ consumedOptions.put(FileBaseSourceOptions.CSV_USE_HEADER_LINE.key(),
true);
+ consumedOptions.put(FileBaseSourceOptions.BINARY_CHUNK_SIZE.key(),
2048);
+
consumedOptions.put(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE.key(),
true);
+ consumedOptions.put(
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START.key(),
"2024-01-01 00:00:00");
+ consumedOptions.put(
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END.key(),
"2024-12-31 23:59:59");
+
consumedOptions.put(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME.key(), true);
+
+ OptionRule optionRule = (new LocalFileSourceFactory()).optionRule();
+
+ for (Map.Entry<String, Object> entry : consumedOptions.entrySet()) {
+ Map<String, Object> config = sourceConfig();
+ config.put(entry.getKey(), entry.getValue());
+
+ Assertions.assertDoesNotThrow(
+ () ->
+ ConfigValidator.validateUnknownKeys(
+ ReadonlyConfig.fromMap(config),
optionRule, "LocalFile"),
+ "validateUnknownKeys should accept declared option: " +
entry.getKey());
+ }
+ }
+
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-obs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/obs/source/ObsFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-obs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/obs/source/ObsFileSourceFactory.java
index 55e068f5d8..23876525b3 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-obs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/obs/source/ObsFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-obs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/obs/source/ObsFileSourceFactory.java
@@ -76,6 +76,8 @@ public class ObsFileSourceFactory implements
TableSourceFactory {
.optional(FileBaseSourceOptions.DATE_FORMAT_LEGACY)
.optional(FileBaseSourceOptions.DATETIME_FORMAT_LEGACY)
.optional(FileBaseSourceOptions.TIME_FORMAT_LEGACY)
+ .optional(FileBaseSourceOptions.COMPRESS_CODEC)
+ .optional(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC)
.optional(FileBaseSourceOptions.NULL_FORMAT)
.optional(FileBaseSourceOptions.FILENAME_EXTENSION)
.optional(FileBaseSourceOptions.READ_COLUMNS)
@@ -94,6 +96,15 @@ public class ObsFileSourceFactory implements
TableSourceFactory {
.optional(FileBaseSourceOptions.QUOTE_CHAR)
.optional(FileBaseSourceOptions.ESCAPE_CHAR)
.optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN)
+ .optional(
+ FileBaseSourceOptions.READ_PARTITIONS,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END,
+ FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)
+ .optional(
+ FileBaseSourceOptions.CSV_USE_HEADER_LINE,
+ FileBaseSourceOptions.BINARY_CHUNK_SIZE,
+ FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-obs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/obs/ObsFileFactoryTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-obs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/obs/ObsFileFactoryTest.java
index dc7c34525a..d87df875ac 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-obs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/obs/ObsFileFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-obs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/obs/ObsFileFactoryTest.java
@@ -17,17 +17,70 @@
package org.apache.seatunnel.connectors.seatunnel.file.obs;
+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.connectors.seatunnel.file.config.FileBaseOptions;
+import
org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.file.obs.sink.ObsFileSinkFactory;
import
org.apache.seatunnel.connectors.seatunnel.file.obs.source.ObsFileSourceFactory;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
public class ObsFileFactoryTest {
@Test
void optionRule() {
- Assertions.assertNotNull((new ObsFileSourceFactory()).optionRule());
+ OptionRule optionRule = (new ObsFileSourceFactory()).optionRule();
+ Assertions.assertNotNull(optionRule);
Assertions.assertNotNull((new ObsFileSinkFactory()).optionRule());
}
+
+ @Test
+ void consumedOptionsAcceptedByValidateUnknownKeys() {
+ // These options are consumed by AbstractReadStrategy /
CsvReadStrategy /
+ // BinaryReadStrategy at runtime. They MUST be declared in the
factory's
+ // optionRule() so that ConfigValidator.validateUnknownKeys (called by
+ // --check / --dry-run at SeaTunnelConfValidateCommand:325) does not
+ // reject them as "unknown option keys".
+ Map<String, Object> consumedOptions = new LinkedHashMap<>();
+ consumedOptions.put(
+ FileBaseSourceOptions.READ_PARTITIONS.key(),
+ Collections.singletonList("year=2024"));
+ consumedOptions.put(FileBaseSourceOptions.CSV_USE_HEADER_LINE.key(),
true);
+ consumedOptions.put(FileBaseSourceOptions.BINARY_CHUNK_SIZE.key(),
2048);
+
consumedOptions.put(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE.key(),
true);
+ consumedOptions.put(
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START.key(),
"2024-01-01 00:00:00");
+ consumedOptions.put(
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END.key(),
"2024-12-31 23:59:59");
+
consumedOptions.put(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME.key(), true);
+ consumedOptions.put(FileBaseSourceOptions.COMPRESS_CODEC.key(),
"gzip");
+
consumedOptions.put(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC.key(), "zip");
+
+ OptionRule optionRule = (new ObsFileSourceFactory()).optionRule();
+
+ for (Map.Entry<String, Object> entry : consumedOptions.entrySet()) {
+ Map<String, Object> config = sourceConfig();
+ config.put(entry.getKey(), entry.getValue());
+
+ Assertions.assertDoesNotThrow(
+ () ->
+ ConfigValidator.validateUnknownKeys(
+ ReadonlyConfig.fromMap(config),
optionRule, "ObsFile"),
+ "validateUnknownKeys should accept declared option: " +
entry.getKey());
+ }
+ }
+
+ private static Map<String, Object> sourceConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put(FileBaseOptions.FILE_PATH.key(), "/source");
+ return config;
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java
index e664d1f82d..634e0ce627 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java
@@ -141,6 +141,15 @@ public class OssFileSourceFactory implements
TableSourceFactory {
FilePostSyncAction.BACKUP,
FileBaseSourceOptions.BACKUP_PATH)
.optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN)
+ .optional(
+ FileBaseSourceOptions.READ_PARTITIONS,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END,
+ FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)
+ .optional(
+ FileBaseSourceOptions.CSV_USE_HEADER_LINE,
+ FileBaseSourceOptions.BINARY_CHUNK_SIZE,
+ FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-oss/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/oss/OssFileFactoryTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-oss/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/oss/OssFileFactoryTest.java
index fe7b37c4d2..0c6042fd72 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-oss/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/oss/OssFileFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-oss/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/oss/OssFileFactoryTest.java
@@ -17,6 +17,8 @@
package org.apache.seatunnel.connectors.seatunnel.file.oss;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import
org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.file.oss.sink.OssFileSinkFactory;
import
org.apache.seatunnel.connectors.seatunnel.file.oss.source.OssFileSourceFactory;
@@ -30,4 +32,33 @@ public class OssFileFactoryTest {
Assertions.assertNotNull((new OssFileSourceFactory()).optionRule());
Assertions.assertNotNull((new OssFileSinkFactory()).optionRule());
}
+
+ @Test
+ void consumedOptionsAreDeclared() {
+ OptionRule optionRule = (new OssFileSourceFactory()).optionRule();
+ Assertions.assertTrue(
+
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.READ_PARTITIONS));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+ .contains(FileBaseSourceOptions.CSV_USE_HEADER_LINE));
+ Assertions.assertTrue(
+
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.BINARY_CHUNK_SIZE));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_START));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_END));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME));
+ }
}
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 02a85e7cd4..03cb29e75c 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
@@ -174,6 +174,15 @@ public class S3FileSourceFactory implements
TableSourceFactory, SupportSourceDry
FilePostSyncAction.BACKUP,
FileBaseSourceOptions.BACKUP_PATH)
.optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN)
+ .optional(
+ FileBaseSourceOptions.READ_PARTITIONS,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END,
+ FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)
+ .optional(
+ FileBaseSourceOptions.CSV_USE_HEADER_LINE,
+ FileBaseSourceOptions.BINARY_CHUNK_SIZE,
+ FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)
.build();
}
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 38d21352d6..3ff80a2184 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
@@ -126,6 +126,35 @@ class S3FileFactoryTest {
Assertions.assertDoesNotThrow(() -> validate(config, optionRule));
}
+ @Test
+ void consumedOptionsAreDeclared() {
+ OptionRule optionRule = new S3FileSourceFactory().optionRule();
+ Assertions.assertTrue(
+
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.READ_PARTITIONS));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+ .contains(FileBaseSourceOptions.CSV_USE_HEADER_LINE));
+ Assertions.assertTrue(
+
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.BINARY_CHUNK_SIZE));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_START));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_END));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME));
+ }
+
private static Map<String, Object> sinkConfig() {
Map<String, Object> config = new HashMap<>();
config.put(S3FileBaseOptions.FILE_PATH.key(), "/tmp/seatunnel");
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java
index 160b260dc5..768c8f75ed 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java
@@ -88,6 +88,8 @@ public class SftpFileSourceFactory implements
TableSourceFactory {
.optional(FileBaseSourceOptions.DATETIME_FORMAT_LEGACY)
.optional(FileBaseSourceOptions.TIME_FORMAT_LEGACY)
.optional(FileBaseSourceOptions.FILE_FILTER_PATTERN)
+ .optional(FileBaseSourceOptions.COMPRESS_CODEC)
+ .optional(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC)
.optional(FileBaseSourceOptions.NULL_FORMAT)
.optional(FileBaseSourceOptions.FILENAME_EXTENSION)
.optional(FileBaseSourceOptions.READ_COLUMNS)
@@ -131,6 +133,15 @@ public class SftpFileSourceFactory implements
TableSourceFactory {
FilePostSyncAction.BACKUP,
FileBaseSourceOptions.BACKUP_PATH)
.optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN)
+ .optional(
+ FileBaseSourceOptions.READ_PARTITIONS,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END,
+ FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)
+ .optional(
+ FileBaseSourceOptions.CSV_USE_HEADER_LINE,
+ FileBaseSourceOptions.BINARY_CHUNK_SIZE,
+ FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/SftpFileFactoryTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/SftpFileFactoryTest.java
index 93d31661ea..272523b688 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/SftpFileFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/SftpFileFactoryTest.java
@@ -32,7 +32,9 @@ import
org.apache.seatunnel.connectors.seatunnel.file.sftp.system.SFTPFileSystem
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.Map;
class SftpFileFactoryTest {
@@ -73,6 +75,12 @@ class SftpFileFactoryTest {
optionRule
.getOptionalOptions()
.contains(FileBaseSourceOptions.RETENTION_CHECK_INTERVAL));
+ Assertions.assertTrue(
+
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.COMPRESS_CODEC));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC));
Assertions.assertTrue(
optionRule.getOptionalOptions().stream()
.anyMatch(option -> "keyfile".equals(option.key())));
@@ -134,6 +142,42 @@ class SftpFileFactoryTest {
hadoopConf.getExtraOptions().get(SFTPFileSystem.FS_SFTP_KEYFILE));
}
+ @Test
+ void consumedOptionsAcceptedByValidateUnknownKeys() {
+ // These options are consumed by AbstractReadStrategy /
CsvReadStrategy /
+ // BinaryReadStrategy at runtime. They MUST be declared in the
factory's
+ // optionRule() so that ConfigValidator.validateUnknownKeys (called by
+ // --check / --dry-run at SeaTunnelConfValidateCommand:325) does not
+ // reject them as "unknown option keys".
+ Map<String, Object> consumedOptions = new LinkedHashMap<>();
+ consumedOptions.put(
+ FileBaseSourceOptions.READ_PARTITIONS.key(),
+ Collections.singletonList("year=2024"));
+ consumedOptions.put(FileBaseSourceOptions.CSV_USE_HEADER_LINE.key(),
true);
+ consumedOptions.put(FileBaseSourceOptions.BINARY_CHUNK_SIZE.key(),
2048);
+
consumedOptions.put(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE.key(),
true);
+ consumedOptions.put(
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START.key(),
"2024-01-01 00:00:00");
+ consumedOptions.put(
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END.key(),
"2024-12-31 23:59:59");
+
consumedOptions.put(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME.key(), true);
+ consumedOptions.put(FileBaseSourceOptions.COMPRESS_CODEC.key(),
"gzip");
+
consumedOptions.put(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC.key(), "zip");
+
+ OptionRule optionRule = (new SftpFileSourceFactory()).optionRule();
+
+ for (Map.Entry<String, Object> entry : consumedOptions.entrySet()) {
+ Map<String, Object> config = sourceConfig();
+ config.put(entry.getKey(), entry.getValue());
+
+ Assertions.assertDoesNotThrow(
+ () ->
+ ConfigValidator.validateUnknownKeys(
+ ReadonlyConfig.fromMap(config),
optionRule, "SftpFile"),
+ "validateUnknownKeys should accept declared option: " +
entry.getKey());
+ }
+ }
+
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-smb/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/smb/source/SmbFileSourceFactory.java
b/seatunnel-connectors-v2/connector-file/connector-file-smb/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/smb/source/SmbFileSourceFactory.java
index 7a7281a77d..c109039f83 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-smb/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/smb/source/SmbFileSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-smb/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/smb/source/SmbFileSourceFactory.java
@@ -89,6 +89,8 @@ public class SmbFileSourceFactory implements
TableSourceFactory {
.optional(FileBaseSourceOptions.DATETIME_FORMAT_LEGACY)
.optional(FileBaseSourceOptions.TIME_FORMAT_LEGACY)
.optional(FileBaseSourceOptions.FILE_FILTER_PATTERN)
+ .optional(FileBaseSourceOptions.COMPRESS_CODEC)
+ .optional(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC)
.optional(FileBaseSourceOptions.NULL_FORMAT)
.optional(FileBaseSourceOptions.FILENAME_EXTENSION)
.optional(FileBaseSourceOptions.READ_COLUMNS)
@@ -132,6 +134,15 @@ public class SmbFileSourceFactory implements
TableSourceFactory {
FilePostSyncAction.BACKUP,
FileBaseSourceOptions.BACKUP_PATH)
.optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN)
+ .optional(
+ FileBaseSourceOptions.READ_PARTITIONS,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START,
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END,
+ FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)
+ .optional(
+ FileBaseSourceOptions.CSV_USE_HEADER_LINE,
+ FileBaseSourceOptions.BINARY_CHUNK_SIZE,
+ FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-smb/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/smb/SmbFileFactoryTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-smb/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/smb/SmbFileFactoryTest.java
index 635174b771..6f50d8651f 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-smb/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/smb/SmbFileFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-smb/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/smb/SmbFileFactoryTest.java
@@ -34,7 +34,9 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -76,6 +78,12 @@ class SmbFileFactoryTest {
optionRule
.getOptionalOptions()
.contains(FileBaseSourceOptions.RETENTION_CHECK_INTERVAL));
+ Assertions.assertTrue(
+
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.COMPRESS_CODEC));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC));
}
@Test
@@ -224,6 +232,42 @@ class SmbFileFactoryTest {
Assertions.assertEquals("smb", hadoopConf.getSchema());
}
+ @Test
+ void consumedOptionsAcceptedByValidateUnknownKeys() {
+ // These options are consumed by AbstractReadStrategy /
CsvReadStrategy /
+ // BinaryReadStrategy at runtime. They MUST be declared in the
factory's
+ // optionRule() so that ConfigValidator.validateUnknownKeys (called by
+ // --check / --dry-run at SeaTunnelConfValidateCommand:325) does not
+ // reject them as "unknown option keys".
+ Map<String, Object> consumedOptions = new LinkedHashMap<>();
+ consumedOptions.put(
+ FileBaseSourceOptions.READ_PARTITIONS.key(),
+ Collections.singletonList("year=2024"));
+ consumedOptions.put(FileBaseSourceOptions.CSV_USE_HEADER_LINE.key(),
true);
+ consumedOptions.put(FileBaseSourceOptions.BINARY_CHUNK_SIZE.key(),
2048);
+
consumedOptions.put(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE.key(),
true);
+ consumedOptions.put(
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_START.key(),
"2024-01-01 00:00:00");
+ consumedOptions.put(
+ FileBaseSourceOptions.FILE_FILTER_MODIFIED_END.key(),
"2024-12-31 23:59:59");
+
consumedOptions.put(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME.key(), true);
+ consumedOptions.put(FileBaseSourceOptions.COMPRESS_CODEC.key(),
"gzip");
+
consumedOptions.put(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC.key(), "zip");
+
+ OptionRule optionRule = (new SmbFileSourceFactory()).optionRule();
+
+ for (Map.Entry<String, Object> entry : consumedOptions.entrySet()) {
+ Map<String, Object> config = sourceConfig();
+ config.put(entry.getKey(), entry.getValue());
+
+ Assertions.assertDoesNotThrow(
+ () ->
+ ConfigValidator.validateUnknownKeys(
+ ReadonlyConfig.fromMap(config),
optionRule, "SmbFile"),
+ "validateUnknownKeys should accept declared option: " +
entry.getKey());
+ }
+ }
+
private static Map<String, Object> sourceConfig() {
Map<String, Object> config = new HashMap<>();
config.put(FileBaseOptions.FILE_PATH.key(), "/source");
