This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-11967-5dc71549adb79a798f4d8a9de64523830d57de39 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 96d8e8d39dfdf25ca497d6414ec7d0bab2796edd Author: Goutam Adwant <[email protected]> AuthorDate: Thu Sep 3 04:58:38 2026 +0000 [Feature][Connector-V2] Add Google Cloud Storage file source (#11967) Signed-off-by: goutamadwant <[email protected]> Signed-off-by: Goutam Adwant <[email protected]> --- config/plugin_config | 1 + docs/en/connectors/changelog/connector-file-gcs.md | 6 + docs/en/connectors/source/GcsFile.md | 138 +++++++++++++++++ docs/zh/connectors/changelog/connector-file-gcs.md | 6 + docs/zh/connectors/source/GcsFile.md | 128 ++++++++++++++++ plugin-mapping.properties | 1 + .../seatunnel/file/config/FileSystemType.java | 3 +- .../connector-file/connector-file-gcs/pom.xml | 69 +++++++++ .../file/gcs/config/GcsFileBaseOptions.java | 49 ++++++ .../file/gcs/config/GcsFileSourceOptions.java} | 27 +--- .../seatunnel/file/gcs/config/GcsHadoopConf.java | 105 +++++++++++++ .../seatunnel/file/gcs/source/GcsFileSource.java | 45 ++++++ .../file/gcs/source/GcsFileSourceFactory.java | 164 +++++++++++++++++++++ .../gcs/source/config/GcsFileSourceConfig.java | 46 ++++++ .../config/MultipleTableGcsFileSourceConfig.java | 39 +++++ .../services/org.apache.hadoop.fs.FileSystem | 16 ++ .../file/gcs/GcsFileSourceFactoryTest.java | 95 ++++++++++++ .../file/gcs/config/GcsHadoopConfTest.java | 130 ++++++++++++++++ seatunnel-connectors-v2/connector-file/pom.xml | 1 + seatunnel-dist/pom.xml | 6 + seatunnel-dist/release-docs/NOTICE | 104 +++++++++++++ .../licenses/LICENSE-gcs-connector.txt | 1 + .../connector-file-gcs-e2e/pom.xml | 58 ++++++++ .../e2e/connector/file/gcs/GcsFileIT.java | 104 +++++++++++++ .../src/test/resources/gcs/data/e2e.json | 2 + .../src/test/resources/gcs/gcs_file_to_assert.conf | 84 +++++++++++ seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml | 1 + .../container/seatunnel/SeaTunnelContainer.java | 24 +++ .../SeaTunnelContainerThreadExemptionTest.java | 24 ++- 29 files changed, 1447 insertions(+), 30 deletions(-) diff --git a/config/plugin_config b/config/plugin_config index a2e8bd3cc7..6c858ed71a 100644 --- a/config/plugin_config +++ b/config/plugin_config @@ -46,6 +46,7 @@ connector-file-local connector-file-oss connector-file-jindo-oss connector-file-s3 +connector-file-gcs connector-file-sftp connector-file-obs connector-file-bos diff --git a/docs/en/connectors/changelog/connector-file-gcs.md b/docs/en/connectors/changelog/connector-file-gcs.md new file mode 100644 index 0000000000..97ff142816 --- /dev/null +++ b/docs/en/connectors/changelog/connector-file-gcs.md @@ -0,0 +1,6 @@ +<details><summary> Change Log </summary> + +| Change | Commit | Version | +|--------|--------|---------| + +</details> diff --git a/docs/en/connectors/source/GcsFile.md b/docs/en/connectors/source/GcsFile.md new file mode 100644 index 0000000000..ecedbfd720 --- /dev/null +++ b/docs/en/connectors/source/GcsFile.md @@ -0,0 +1,138 @@ +import ChangeLog from '../changelog/connector-file-gcs.md'; + +# GcsFile + +> Google Cloud Storage file source connector + +## Support Those Engines + +> Spark<br/> +> Flink<br/> +> SeaTunnel Zeta<br/> + +## Key Features + +- [x] [batch](../../introduction/concepts/connector-v2-features.md) +- [x] [exactly-once](../../introduction/concepts/connector-v2-features.md) +- [x] [column projection](../../introduction/concepts/connector-v2-features.md) +- [x] [parallelism](../../introduction/concepts/connector-v2-features.md) +- [x] [multimodal](../../introduction/concepts/connector-v2-features.md#multimodal) +- [x] multiple table source +- [x] file formats: `text`, `csv`, `parquet`, `orc`, `json`, `excel`, `xml`, `binary`, `markdown`, and `pdf` + +## Description + +Reads files from Google Cloud Storage through the Google Cloud Storage connector for Hadoop. The +connector reuses SeaTunnel's file source implementation for format parsing, schema discovery, +projection, splitting, and multiple-table jobs. + +Set `bucket` to a bucket URI such as `gs://my-bucket`. Set `path` to the object or prefix inside that +bucket, such as `/data/orders`. Do not include an object path in `bucket`. + +## Dependency + +The connector uses `com.google.cloud.bigdataoss:gcs-connector:hadoop3-2.2.33:shaded`, which is +Apache License 2.0 software and targets Java 8. The shaded GCS Hadoop library is packaged in the +`connector-file-gcs` connector JAR. Spark and Flink deployments must provide a compatible Hadoop 3 +runtime on every driver and worker. + +## Authentication + +The connector supports these authentication modes: + +1. **Application Default Credentials (ADC):** omit `service_account_key_file`. The Hadoop GCS + connector discovers credentials from `GOOGLE_APPLICATION_CREDENTIALS` or the service account + attached to the Google Cloud runtime. +2. **Service-account JSON file:** set `service_account_key_file` to a local path that exists at the + same location on every node that reads GCS. + +The explicit `service_account_key_file` option takes precedence over the corresponding entry in +`hadoop_gcs_properties`. + +## Options + +| Name | Type | Required | Default | Description | +|------|------|----------|---------|-------------| +| path | string | yes | - | Object or prefix path inside `bucket`, for example `/data/orders`. | +| file_format_type | string | yes | - | File format: `text`, `csv`, `parquet`, `orc`, `json`, `excel`, `xml`, `binary`, `markdown`, or `pdf`. | +| bucket | string | yes | - | GCS bucket URI, for example `gs://my-bucket`. | +| service_account_key_file | string | no | - | Service-account JSON key file on every worker. When omitted, ADC is used. | +| hadoop_gcs_properties | map | no | - | Additional `fs.gs.*` Hadoop properties. Explicit connector options take precedence. | +| schema | config | conditional | - | Required for `text`, `json`, `excel`, `csv`, and `xml`. See [Schema Feature](../../introduction/concepts/schema-feature.md). | +| read_columns | list | no | - | Columns to project from the source. | +| field_delimiter | string | no | `\001` for text, `,` for CSV | Field delimiter for text and CSV files. `delimiter` is an alias. | +| row_delimiter | string | no | `\n` | Row delimiter for text files. | +| skip_header_row_number | long | no | `0` | Number of initial text or CSV rows to skip. | +| encoding | string | no | `UTF-8` | Character encoding for text, JSON, CSV, and XML files. | +| parse_partition_from_path | boolean | no | `true` | Adds partition values parsed from paths such as `/year=2026/month=08`. | +| recursive_file_scan | boolean | no | `true` | Whether to scan subdirectories recursively. | +| file_filter_pattern | string | no | - | File name filter pattern. | +| filename_extension | string | no | - | File extension filter, for example `csv` or `.json`. | +| compress_codec | string | no | `none` | Compression codec for a single compressed file. | +| archive_compress_codec | string | no | `none` | Archive compression codec. | +| 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. | +| quote_char | string | no | `"` | CSV quote character. | +| escape_char | string | no | - | CSV escape character. | +| sheet_name | string | no | - | Excel worksheet to read. | +| excel_engine | string | no | `POI` | Excel reader: `POI` or `EasyExcel`. | +| 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. | +| 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. | +| sync_mode | string | no | `full` | `full` or `update`. Update mode currently supports binary format only. | +| target_path | string | conditional | - | Required for `sync_mode=update`; used to compare objects by relative path. | +| target_hadoop_conf | map | no | - | Hadoop configuration for the comparison target. | +| update_strategy | string | no | `distcp` | Update comparison strategy: `distcp` or `strict`. | +| compare_mode | string | no | `len_mtime` | `len_mtime` or `checksum`; checksum requires strict strategy. | +| update_compare_parallelism | int | no | `8` | Parallelism for target metadata lookups, from 1 through 64. | +| update_compare_bulk_threshold | int | no | `0` | Positive values enable bulk directory listing at the threshold; `0` disables it. | +| post_sync_action | string | no | `none` | Post-checkpoint action for continuous discovery: `none`, `delete`, or `backup`. | +| 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. | +| common-options | | no | - | See [Source Common Options](../common-options/source-common-options.md). | + +## Example + +### Read Parquet With ADC + +```hocon +source { + GcsFile { + bucket = "gs://my-bucket" + path = "/warehouse/orders" + file_format_type = "parquet" + } +} +``` + +### Read CSV With a Service Account + +```hocon +source { + GcsFile { + bucket = "gs://my-bucket" + path = "/landing/customers" + file_format_type = "csv" + service_account_key_file = "/opt/seatunnel/keys/gcs-reader.json" + skip_header_row_number = 1 + schema { + fields { + id = long + name = string + } + } + hadoop_gcs_properties = { + "fs.gs.project.id" = "my-project" + } + } +} +``` + +## Changelog + +<ChangeLog /> diff --git a/docs/zh/connectors/changelog/connector-file-gcs.md b/docs/zh/connectors/changelog/connector-file-gcs.md new file mode 100644 index 0000000000..97ff142816 --- /dev/null +++ b/docs/zh/connectors/changelog/connector-file-gcs.md @@ -0,0 +1,6 @@ +<details><summary> Change Log </summary> + +| Change | Commit | Version | +|--------|--------|---------| + +</details> diff --git a/docs/zh/connectors/source/GcsFile.md b/docs/zh/connectors/source/GcsFile.md new file mode 100644 index 0000000000..dd58c949cd --- /dev/null +++ b/docs/zh/connectors/source/GcsFile.md @@ -0,0 +1,128 @@ +import ChangeLog from '../changelog/connector-file-gcs.md'; + +# GcsFile + +> Google Cloud Storage 文件 Source 连接器 + +## 支持的引擎 + +> Spark<br/> +> Flink<br/> +> SeaTunnel Zeta<br/> + +## 主要特性 + +- [x] [批处理](../../introduction/concepts/connector-v2-features.md) +- [x] [精确一次](../../introduction/concepts/connector-v2-features.md) +- [x] [列投影](../../introduction/concepts/connector-v2-features.md) +- [x] [并行度](../../introduction/concepts/connector-v2-features.md) +- [x] [多模态](../../introduction/concepts/connector-v2-features.md#multimodal) +- [x] 多表 Source +- [x] 文件格式：`text`、`csv`、`parquet`、`orc`、`json`、`excel`、`xml`、`binary`、`markdown` 和 `pdf` + +## 描述 + +通过 Google Cloud Storage Hadoop 连接器读取 GCS 文件。格式解析、Schema 发现、列投影、文件切分和多表任务复用 SeaTunnel File Source 的现有实现。 + +`bucket` 必须是 `gs://my-bucket` 形式的存储桶 URI。`path` 是存储桶内的对象或前缀，例如 `/data/orders`。不要把对象路径写入 `bucket`。 + +## 依赖 + +本连接器使用 `com.google.cloud.bigdataoss:gcs-connector:hadoop3-2.2.33:shaded`。该依赖采用 Apache License 2.0，并以 Java 8 为目标版本。shaded GCS Hadoop 库已打包到 `connector-file-gcs` 连接器 JAR 中。Spark 和 Flink 部署必须在驱动节点和所有工作节点提供兼容的 Hadoop 3 运行环境。 + +## 认证 + +连接器支持以下认证方式： + +1. **应用默认凭据（ADC）**：不配置 `service_account_key_file`。Hadoop GCS 连接器会从 `GOOGLE_APPLICATION_CREDENTIALS` 或 Google Cloud 运行环境绑定的服务账号获取凭据。 +2. **服务账号 JSON 文件**：配置 `service_account_key_file`。该本地路径必须以相同位置存在于每个读取 GCS 的节点上。 + +显式配置的 `service_account_key_file` 优先于 `hadoop_gcs_properties` 中的同名 Hadoop 属性。 + +## 配置项 + +| 名称 | 类型 | 是否必填 | 默认值 | 描述 | +|------|------|----------|--------|------| +| path | string | 是 | - | `bucket` 内的对象或前缀路径，例如 `/data/orders`。 | +| file_format_type | string | 是 | - | 文件格式：`text`、`csv`、`parquet`、`orc`、`json`、`excel`、`xml`、`binary`、`markdown` 或 `pdf`。 | +| bucket | string | 是 | - | GCS 存储桶 URI，例如 `gs://my-bucket`。 | +| service_account_key_file | string | 否 | - | 每个工作节点上的服务账号 JSON 文件。省略时使用 ADC。 | +| hadoop_gcs_properties | map | 否 | - | 额外的 `fs.gs.*` Hadoop 属性。显式连接器配置优先。 | +| schema | config | 条件必填 | - | `text`、`json`、`excel`、`csv` 和 `xml` 格式需要配置。参见 [Schema 功能](../../introduction/concepts/schema-feature.md)。 | +| read_columns | list | 否 | - | 从数据源投影的列。 | +| field_delimiter | string | 否 | text 为 `\001`，CSV 为 `,` | text 和 CSV 的字段分隔符，`delimiter` 是其别名。 | +| row_delimiter | string | 否 | `\n` | text 文件的行分隔符。 | +| skip_header_row_number | long | 否 | `0` | 跳过 text 或 CSV 文件开头的行数。 | +| encoding | string | 否 | `UTF-8` | text、JSON、CSV 和 XML 的字符编码。 | +| parse_partition_from_path | boolean | 否 | `true` | 从 `/year=2026/month=08` 形式的路径解析分区值。 | +| recursive_file_scan | boolean | 否 | `true` | 是否递归扫描子目录。 | +| file_filter_pattern | string | 否 | - | 文件名过滤模式。 | +| filename_extension | string | 否 | - | 文件扩展名过滤，例如 `csv` 或 `.json`。 | +| compress_codec | string | 否 | `none` | 单个压缩文件的压缩编码。 | +| archive_compress_codec | string | 否 | `none` | 归档压缩编码。 | +| enable_file_split | boolean | 否 | `false` | 为未压缩的 text、CSV、JSON 和 Parquet 文件启用逻辑切分。 | +| file_split_size | long | 条件必填 | `134217728` | `enable_file_split=true` 时的切分大小，单位为字节。 | +| null_format | string | 否 | - | 表示空值的文本。 | +| quote_char | string | 否 | `"` | CSV 引号字符。 | +| escape_char | string | 否 | - | CSV 转义字符。 | +| sheet_name | string | 否 | - | 要读取的 Excel 工作表。 | +| excel_engine | string | 否 | `POI` | Excel 读取器：`POI` 或 `EasyExcel`。 | +| poi_excel_max_file_size | long | 否 | `52428800` | POI 引擎可读取的最大 Excel 文件字节数。 | +| xml_row_tag | string | 条件必填 | - | 表示一行数据的 XML 元素。 | +| xml_use_attr_format | boolean | 条件必填 | - | 是否从 XML 属性读取值。 | +| discovery_mode | string | 否 | `once` | `once` 或 `continuous`。连续发现当前要求 update 同步和 binary 格式。 | +| scan_interval | string | 否 | `10S` | 连续发现的轮询间隔。 | +| start_mode | string | 否 | `earliest` | `earliest` 处理已有文件，`latest` 从后续变更开始。 | +| sync_mode | string | 否 | `full` | `full` 或 `update`。update 当前仅支持 binary 格式。 | +| target_path | string | 条件必填 | - | `sync_mode=update` 时必填，用于按相对路径比较对象。 | +| target_hadoop_conf | map | 否 | - | 比较目标文件系统的 Hadoop 配置。 | +| update_strategy | string | 否 | `distcp` | 更新比较策略：`distcp` 或 `strict`。 | +| compare_mode | string | 否 | `len_mtime` | `len_mtime` 或 `checksum`；checksum 要求 strict 策略。 | +| update_compare_parallelism | int | 否 | `8` | 目标元数据查询并行度，范围为 1 到 64。 | +| update_compare_bulk_threshold | int | 否 | `0` | 正数表示达到阈值后使用批量目录列举；`0` 表示禁用。 | +| post_sync_action | string | 否 | `none` | 连续发现对象完成 checkpoint 后的动作：`none`、`delete` 或 `backup`。 | +| backup_path | string | 条件必填 | - | `post_sync_action=backup` 时必填，且不能与源路径重叠。 | +| retention_max_age | string | 否 | - | SeaTunnel 备份对象的最大保留时间。 | +| retention_check_interval | string | 否 | `1H` | 备份保留扫描间隔。 | +| common-options | | 否 | - | 参见 [Source 通用配置](../common-options/source-common-options.md)。 | + +## 示例 + +### 使用 ADC 读取 Parquet + +```hocon +source { + GcsFile { + bucket = "gs://my-bucket" + path = "/warehouse/orders" + file_format_type = "parquet" + } +} +``` + +### 使用服务账号读取 CSV + +```hocon +source { + GcsFile { + bucket = "gs://my-bucket" + path = "/landing/customers" + file_format_type = "csv" + service_account_key_file = "/opt/seatunnel/keys/gcs-reader.json" + skip_header_row_number = 1 + schema { + fields { + id = long + name = string + } + } + hadoop_gcs_properties = { + "fs.gs.project.id" = "my-project" + } + } +} +``` + +## Changelog + +<ChangeLog /> diff --git a/plugin-mapping.properties b/plugin-mapping.properties index d0253dcb4d..e04731eb67 100644 --- a/plugin-mapping.properties +++ b/plugin-mapping.properties @@ -83,6 +83,7 @@ seatunnel.sink.Iceberg = connector-iceberg seatunnel.source.InfluxDB = connector-influxdb seatunnel.source.S3File = connector-file-s3 seatunnel.sink.S3File = connector-file-s3 +seatunnel.source.GcsFile = connector-file-gcs seatunnel.source.AmazonDynamodb = connector-amazondynamodb seatunnel.sink.AmazonDynamodb = connector-amazondynamodb seatunnel.source.AzureCosmosDB = connector-azurecosmosdb diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileSystemType.java b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileSystemType.java index 19c67151c0..0b74520c72 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileSystemType.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileSystemType.java @@ -29,7 +29,8 @@ public enum FileSystemType implements Serializable { SFTP("SftpFile"), S3("S3File"), OBS("ObsFile"), - BOS("BosFile"); + BOS("BosFile"), + GCS("GcsFile"); private final String fileSystemPluginName; diff --git a/seatunnel-connectors-v2/connector-file/connector-file-gcs/pom.xml b/seatunnel-connectors-v2/connector-file/connector-file-gcs/pom.xml new file mode 100644 index 0000000000..4d44fe58ca --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-gcs/pom.xml @@ -0,0 +1,69 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + + 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. + +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-file</artifactId> + <version>${revision}</version> + </parent> + + <artifactId>connector-file-gcs</artifactId> + <name>SeaTunnel : Connectors V2 : File : GCS</name> + + <properties> + <gcs-connector.version>hadoop3-2.2.33</gcs-connector.version> + </properties> + + <dependencies> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-file-base</artifactId> + <version>${project.version}</version> + </dependency> + + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>seatunnel-shade-hadoop3-uber</artifactId> + <scope>provided</scope> + <exclusions> + <exclusion> + <groupId>org.apache.avro</groupId> + <artifactId>avro</artifactId> + </exclusion> + </exclusions> + </dependency> + + <dependency> + <groupId>com.google.cloud.bigdataoss</groupId> + <artifactId>gcs-connector</artifactId> + <version>${gcs-connector.version}</version> + <classifier>shaded</classifier> + <exclusions> + <exclusion> + <groupId>*</groupId> + <artifactId>*</artifactId> + </exclusion> + </exclusions> + </dependency> + </dependencies> + +</project> diff --git a/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/config/GcsFileBaseOptions.java b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/config/GcsFileBaseOptions.java new file mode 100644 index 0000000000..98fb0e5004 --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/config/GcsFileBaseOptions.java @@ -0,0 +1,49 @@ +/* + * 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.gcs.config; + +import org.apache.seatunnel.api.configuration.Option; +import org.apache.seatunnel.api.configuration.Options; +import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions; + +import java.util.Map; + +public class GcsFileBaseOptions extends FileBaseSourceOptions { + + public static final Option<String> BUCKET = + Options.key("bucket") + .stringType() + .noDefaultValue() + .withDescription("Google Cloud Storage bucket URI, for example gs://my-bucket"); + + public static final Option<String> SERVICE_ACCOUNT_KEY_FILE = + Options.key("service_account_key_file") + .stringType() + .noDefaultValue() + .withDescription( + "Path to a service account JSON key file available on every worker. " + + "When omitted, the GCS connector uses Application Default Credentials"); + + public static final Option<Map<String, String>> GCS_PROPERTIES = + Options.key("hadoop_gcs_properties") + .mapType() + .noDefaultValue() + .withDescription( + "Additional Hadoop GCS connector properties. Explicit connector " + + "options take precedence over entries in this map"); +} diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileSystemType.java b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/config/GcsFileSourceOptions.java similarity index 56% copy from seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileSystemType.java copy to seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/config/GcsFileSourceOptions.java index 19c67151c0..8a8b42199f 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileSystemType.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/config/GcsFileSourceOptions.java @@ -15,29 +15,6 @@ * limitations under the License. */ -package org.apache.seatunnel.connectors.seatunnel.file.config; +package org.apache.seatunnel.connectors.seatunnel.file.gcs.config; -import java.io.Serializable; - -public enum FileSystemType implements Serializable { - HDFS("HdfsFile"), - LOCAL("LocalFile"), - OSS("OssFile"), - OSS_JINDO("OssJindoFile"), - COS("CosFile"), - FTP("FtpFile"), - SFTP("SftpFile"), - S3("S3File"), - OBS("ObsFile"), - BOS("BosFile"); - - private final String fileSystemPluginName; - - FileSystemType(String fileSystemPluginName) { - this.fileSystemPluginName = fileSystemPluginName; - } - - public String getFileSystemPluginName() { - return fileSystemPluginName; - } -} +public class GcsFileSourceOptions extends GcsFileBaseOptions {} diff --git a/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/config/GcsHadoopConf.java b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/config/GcsHadoopConf.java new file mode 100644 index 0000000000..291147c35d --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/config/GcsHadoopConf.java @@ -0,0 +1,105 @@ +/* + * 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.gcs.config; + +import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf; + +import java.net.URI; +import java.net.URISyntaxException; +import java.util.HashMap; +import java.util.Map; + +/** Hadoop filesystem configuration for Google Cloud Storage. */ +public class GcsHadoopConf extends HadoopConf { + + static final String GCS_FILESYSTEM_IMPLEMENTATION = + "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"; + static final String GCS_SERVICE_ACCOUNT_KEY_FILE = "fs.gs.auth.service.account.json.keyfile"; + private static final String SCHEMA = "gs"; + + private GcsHadoopConf(String bucket) { + super(bucket); + } + + /** Builds a GCS Hadoop configuration from connector options. */ + public static HadoopConf buildWithReadonlyConfig(ReadonlyConfig config) { + String bucket = config.get(GcsFileBaseOptions.BUCKET); + validateBucket(bucket); + + Map<String, String> properties = new HashMap<>(); + config.getOptional(GcsFileBaseOptions.GCS_PROPERTIES).ifPresent(properties::putAll); + + config.getOptional(GcsFileBaseOptions.SERVICE_ACCOUNT_KEY_FILE) + .ifPresent(keyFile -> configureServiceAccount(properties, keyFile)); + + GcsHadoopConf hadoopConf = new GcsHadoopConf(bucket); + hadoopConf.setExtraOptions(properties); + return hadoopConf; + } + + @Override + public String getFsHdfsImpl() { + return GCS_FILESYSTEM_IMPLEMENTATION; + } + + @Override + public String getSchema() { + return SCHEMA; + } + + private static void validateBucket(String bucket) { + if (StringUtils.isBlank(bucket)) { + throw invalidBucket(bucket); + } + try { + URI uri = new URI(bucket); + String path = uri.getPath(); + if (!SCHEMA.equals(uri.getScheme()) + || StringUtils.isBlank(uri.getAuthority()) + || uri.getUserInfo() != null + || uri.getPort() != -1 + || uri.getQuery() != null + || uri.getFragment() != null + || (StringUtils.isNotEmpty(path) && !"/".equals(path))) { + throw invalidBucket(bucket); + } + } catch (URISyntaxException e) { + throw invalidBucket(bucket); + } + } + + private static IllegalArgumentException invalidBucket(String bucket) { + return new IllegalArgumentException( + String.format( + "The GCS bucket must be a bucket URI such as 'gs://my-bucket', but was '%s'. " + + "Configure object paths with the 'path' option.", + bucket)); + } + + private static void configureServiceAccount( + Map<String, String> properties, String serviceAccountKeyFile) { + if (StringUtils.isBlank(serviceAccountKeyFile)) { + throw new IllegalArgumentException( + "The GCS service_account_key_file option must not be blank"); + } + properties.put(GCS_SERVICE_ACCOUNT_KEY_FILE, serviceAccountKeyFile); + } +} diff --git a/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/GcsFileSource.java b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/GcsFileSource.java new file mode 100644 index 0000000000..518dde49b6 --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/GcsFileSource.java @@ -0,0 +1,45 @@ +/* + * 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.gcs.source; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.connectors.seatunnel.file.config.FileSystemType; +import org.apache.seatunnel.connectors.seatunnel.file.gcs.source.config.MultipleTableGcsFileSourceConfig; +import org.apache.seatunnel.connectors.seatunnel.file.source.BaseMultipleTableFileSource; + +import java.util.List; + +/** File source implementation for objects stored in Google Cloud Storage. */ +public class GcsFileSource extends BaseMultipleTableFileSource { + + /** Creates a GCS file source for the configured catalog tables. */ + public GcsFileSource( + ReadonlyConfig readonlyConfig, List<CatalogTable> catalogTablesFromConfig) { + this(new MultipleTableGcsFileSourceConfig(readonlyConfig, catalogTablesFromConfig)); + } + + private GcsFileSource(MultipleTableGcsFileSourceConfig sourceConfig) { + super(sourceConfig, initFileSplitStrategy(sourceConfig)); + } + + @Override + public String getPluginName() { + return FileSystemType.GCS.getFileSystemPluginName(); + } +} 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 new file mode 100644 index 0000000000..9f50be47ac --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/GcsFileSourceFactory.java @@ -0,0 +1,164 @@ +/* + * 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.gcs.source; + +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.connector.TableSource; +import org.apache.seatunnel.api.table.factory.Factory; +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; +import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat; +import org.apache.seatunnel.connectors.seatunnel.file.config.FilePostSyncAction; +import org.apache.seatunnel.connectors.seatunnel.file.config.FileSyncMode; +import org.apache.seatunnel.connectors.seatunnel.file.config.FileSystemType; +import org.apache.seatunnel.connectors.seatunnel.file.gcs.config.GcsFileSourceOptions; + +import com.google.auto.service.AutoService; + +import java.io.Serializable; +import java.util.Arrays; + +/** Creates Google Cloud Storage file sources from table factory configuration. */ +@AutoService(Factory.class) +public class GcsFileSourceFactory implements TableSourceFactory { + + @Override + public String factoryIdentifier() { + return FileSystemType.GCS.getFileSystemPluginName(); + } + + @Override + public <T, SplitT extends SourceSplit, StateT extends Serializable> + TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) { + return () -> + (SeaTunnelSource<T, SplitT, StateT>) + new GcsFileSource(context.getOptions(), discoverTableSchemas(context)); + } + + @Override + public OptionRule optionRule() { + return OptionRule.builder() + .required(GcsFileSourceOptions.FILE_PATH) + .required(GcsFileSourceOptions.FILE_FORMAT_TYPE) + .required(GcsFileSourceOptions.BUCKET) + .optional( + GcsFileSourceOptions.SERVICE_ACCOUNT_KEY_FILE, + GcsFileSourceOptions.GCS_PROPERTIES) + .conditional( + FileBaseSourceOptions.FILE_FORMAT_TYPE, + FileFormat.TEXT, + FileBaseSourceOptions.ROW_DELIMITER, + FileBaseSourceOptions.FIELD_DELIMITER, + FileBaseSourceOptions.SKIP_HEADER_ROW_NUMBER) + .conditional( + FileBaseSourceOptions.FILE_FORMAT_TYPE, + FileFormat.XML, + FileBaseSourceOptions.XML_ROW_TAG, + FileBaseSourceOptions.XML_USE_ATTR_FORMAT) + .conditional( + FileBaseSourceOptions.FILE_FORMAT_TYPE, + FileFormat.CSV, + FileBaseSourceOptions.SKIP_HEADER_ROW_NUMBER) + .conditional( + FileBaseSourceOptions.FILE_FORMAT_TYPE, + Arrays.asList( + FileFormat.TEXT, + FileFormat.JSON, + FileFormat.EXCEL, + FileFormat.CSV, + FileFormat.XML), + ConnectorCommonOptions.SCHEMA) + .conditional( + FileBaseSourceOptions.FILE_FORMAT_TYPE, + Arrays.asList( + FileFormat.TEXT, FileFormat.JSON, FileFormat.CSV, FileFormat.XML), + FileBaseSourceOptions.ENCODING) + .conditional( + FileBaseSourceOptions.FILE_FORMAT_TYPE, + Arrays.asList( + FileFormat.TEXT, + FileFormat.JSON, + FileFormat.CSV, + FileFormat.PARQUET), + FileBaseSourceOptions.ENABLE_FILE_SPLIT) + .conditional( + FileBaseSourceOptions.ENABLE_FILE_SPLIT, + Boolean.TRUE, + FileBaseSourceOptions.FILE_SPLIT_SIZE) + .optional(FileBaseSourceOptions.PARSE_PARTITION_FROM_PATH) + .optional(FileBaseSourceOptions.DATE_FORMAT_LEGACY) + .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) + .optional( + FileBaseSourceOptions.SHEET_NAME, + FileBaseSourceOptions.EXCEL_ENGINE, + FileBaseSourceOptions.POI_EXCEL_MAX_FILE_SIZE) + .conditional( + FileBaseSourceOptions.FILE_FORMAT_TYPE, + FileFormat.MARKDOWN, + FileBaseSourceOptions.MARKDOWN_RAG_METADATA_ENABLED) + .conditional( + FileBaseSourceOptions.FILE_FORMAT_TYPE, + FileFormat.PDF, + FileBaseSourceOptions.PDF_RAG_METADATA_ENABLED) + .optional(FileBaseSourceOptions.QUOTE_CHAR) + .optional(FileBaseSourceOptions.ESCAPE_CHAR) + .optional(ConnectorCommonOptions.METALAKE_TYPE) + .optional( + FileBaseSourceOptions.DISCOVERY_MODE, + FileBaseSourceOptions.SCAN_INTERVAL, + FileBaseSourceOptions.START_MODE) + .optional( + FileBaseSourceOptions.SYNC_MODE, + FileBaseSourceOptions.TARGET_HADOOP_CONF, + FileBaseSourceOptions.UPDATE_STRATEGY, + FileBaseSourceOptions.COMPARE_MODE, + FileBaseSourceOptions.UPDATE_COMPARE_PARALLELISM, + FileBaseSourceOptions.UPDATE_COMPARE_BULK_THRESHOLD) + .optional( + FileBaseSourceOptions.POST_SYNC_ACTION, + FileBaseSourceOptions.BACKUP_PATH, + FileBaseSourceOptions.RETENTION_MAX_AGE, + FileBaseSourceOptions.RETENTION_CHECK_INTERVAL) + .conditional( + FileBaseSourceOptions.SYNC_MODE, + FileSyncMode.UPDATE, + FileBaseSourceOptions.TARGET_PATH) + .conditional( + FileBaseSourceOptions.POST_SYNC_ACTION, + FilePostSyncAction.BACKUP, + FileBaseSourceOptions.BACKUP_PATH) + .optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN) + .build(); + } + + @Override + public Class<? extends SeaTunnelSource> getSourceClass() { + return GcsFileSource.class; + } +} diff --git a/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/config/GcsFileSourceConfig.java b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/config/GcsFileSourceConfig.java new file mode 100644 index 0000000000..e5cb57939c --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/config/GcsFileSourceConfig.java @@ -0,0 +1,46 @@ +/* + * 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.gcs.source.config; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.connectors.seatunnel.file.config.BaseFileSourceConfig; +import org.apache.seatunnel.connectors.seatunnel.file.config.FileSystemType; +import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf; +import org.apache.seatunnel.connectors.seatunnel.file.gcs.config.GcsHadoopConf; + +/** Source configuration that supplies the Google Cloud Storage filesystem settings. */ +public class GcsFileSourceConfig extends BaseFileSourceConfig { + + private static final long serialVersionUID = 1L; + + /** Creates a source configuration for a single catalog table. */ + public GcsFileSourceConfig(ReadonlyConfig readonlyConfig, CatalogTable catalogTableFromConfig) { + super(readonlyConfig, catalogTableFromConfig); + } + + @Override + public HadoopConf getHadoopConfig() { + return GcsHadoopConf.buildWithReadonlyConfig(getBaseFileSourceConfig()); + } + + @Override + public String getPluginName() { + return FileSystemType.GCS.getFileSystemPluginName(); + } +} diff --git a/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/config/MultipleTableGcsFileSourceConfig.java b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/config/MultipleTableGcsFileSourceConfig.java new file mode 100644 index 0000000000..24089d2d69 --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/config/MultipleTableGcsFileSourceConfig.java @@ -0,0 +1,39 @@ +/* + * 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.gcs.source.config; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.connectors.seatunnel.file.config.BaseFileSourceConfig; +import org.apache.seatunnel.connectors.seatunnel.file.config.BaseMultipleTableFileSourceConfig; + +import java.util.List; + +public class MultipleTableGcsFileSourceConfig extends BaseMultipleTableFileSourceConfig { + + public MultipleTableGcsFileSourceConfig( + ReadonlyConfig rootConfig, List<CatalogTable> catalogTablesFromConfig) { + super(rootConfig, catalogTablesFromConfig); + } + + @Override + public BaseFileSourceConfig getBaseSourceConfig( + ReadonlyConfig readonlyConfig, CatalogTable catalogTableFromConfig) { + return new GcsFileSourceConfig(readonlyConfig, catalogTableFromConfig); + } +} diff --git a/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/resources/META-INF/services/org.apache.hadoop.fs.FileSystem b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/resources/META-INF/services/org.apache.hadoop.fs.FileSystem new file mode 100644 index 0000000000..1668b432e9 --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/resources/META-INF/services/org.apache.hadoop.fs.FileSystem @@ -0,0 +1,16 @@ +# 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. + +com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem 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 new file mode 100644 index 0000000000..677b3f0ebd --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/GcsFileSourceFactoryTest.java @@ -0,0 +1,95 @@ +/* + * 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.gcs; + +import org.apache.seatunnel.api.configuration.Option; +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.configuration.util.ConfigValidator; +import org.apache.seatunnel.api.configuration.util.OptionRule; +import org.apache.seatunnel.api.configuration.util.OptionValidationException; +import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions; +import org.apache.seatunnel.connectors.seatunnel.file.gcs.config.GcsFileSourceOptions; +import org.apache.seatunnel.connectors.seatunnel.file.gcs.source.GcsFileSource; +import org.apache.seatunnel.connectors.seatunnel.file.gcs.source.GcsFileSourceFactory; + +import org.junit.jupiter.api.Test; + +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; + +class GcsFileSourceFactoryTest { + + @Test + void shouldIdentifyGcsFileSource() { + GcsFileSourceFactory factory = new GcsFileSourceFactory(); + + assertEquals("GcsFile", factory.factoryIdentifier()); + assertEquals(GcsFileSource.class, factory.getSourceClass()); + } + + @Test + void shouldRequireOnlyCoreGcsSourceOptionsForBinaryFiles() { + OptionRule optionRule = new GcsFileSourceFactory().optionRule(); + Map<String, Object> config = sourceConfig(); + + assertDoesNotThrow(() -> validate(config, optionRule)); + + config.remove(GcsFileSourceOptions.BUCKET.key()); + assertThrows(OptionValidationException.class, () -> validate(config, optionRule)); + } + + @Test + void shouldExposeAuthenticationAndFileBaseOptions() { + OptionRule optionRule = new GcsFileSourceFactory().optionRule(); + + assertTrue( + optionRule + .getOptionalOptions() + .contains(GcsFileSourceOptions.SERVICE_ACCOUNT_KEY_FILE)); + assertTrue(optionRule.getOptionalOptions().contains(GcsFileSourceOptions.GCS_PROPERTIES)); + assertTrue(optionRule.getOptionalOptions().contains(FileBaseSourceOptions.DISCOVERY_MODE)); + assertTrue(optionRule.getOptionalOptions().contains(FileBaseSourceOptions.SYNC_MODE)); + assertTrue(optionRuleContains(optionRule, FileBaseSourceOptions.ENABLE_FILE_SPLIT)); + assertTrue(optionRuleContains(optionRule, FileBaseSourceOptions.FILE_SPLIT_SIZE)); + } + + private static Map<String, Object> sourceConfig() { + Map<String, Object> config = new HashMap<>(); + config.put(GcsFileSourceOptions.FILE_PATH.key(), "/source"); + config.put(GcsFileSourceOptions.FILE_FORMAT_TYPE.key(), "binary"); + config.put(GcsFileSourceOptions.BUCKET.key(), "gs://test-bucket"); + return config; + } + + private static boolean optionRuleContains(OptionRule optionRule, Option<?> option) { + if (optionRule.getOptionalOptions().contains(option)) { + return true; + } + return optionRule.getRequiredOptions().stream() + .anyMatch(requiredOption -> requiredOption.getOptions().contains(option)); + } + + private static void validate(Map<String, Object> config, OptionRule optionRule) { + ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(optionRule); + } +} diff --git a/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/config/GcsHadoopConfTest.java b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/config/GcsHadoopConfTest.java new file mode 100644 index 0000000000..5b16044120 --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/config/GcsHadoopConfTest.java @@ -0,0 +1,130 @@ +/* + * 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.gcs.config; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; + +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +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; + +class GcsHadoopConfTest { + + @Test + void shouldConfigureGsFileSystemAndUseDefaultCredentialsWhenKeyFileIsAbsent() throws Exception { + HadoopConf hadoopConf = build(Collections.emptyMap()); + + assertEquals("gs", hadoopConf.getSchema()); + assertEquals("gs://test-bucket", hadoopConf.getHdfsNameKey()); + assertEquals(GcsHadoopConf.GCS_FILESYSTEM_IMPLEMENTATION, hadoopConf.getFsHdfsImpl()); + assertFalse( + hadoopConf + .getExtraOptions() + .containsKey(GcsHadoopConf.GCS_SERVICE_ACCOUNT_KEY_FILE)); + + Configuration configuration = hadoopConf.toConfiguration(); + hadoopConf.setExtraOptionsForConfiguration(configuration); + assertEquals("gs://test-bucket", configuration.get("fs.defaultFS")); + assertEquals(GcsHadoopConf.GCS_FILESYSTEM_IMPLEMENTATION, configuration.get("fs.gs.impl")); + assertTrue(configuration.getBoolean("fs.gs.impl.disable.cache", false)); + assertEquals( + GcsHadoopConf.GCS_FILESYSTEM_IMPLEMENTATION, + FileSystem.getFileSystemClass("gs", configuration).getName()); + } + + @Test + void shouldConfigureServiceAccountJsonKeyFile() { + Map<String, Object> options = new HashMap<>(); + options.put(GcsFileBaseOptions.SERVICE_ACCOUNT_KEY_FILE.key(), "/keys/gcs.json"); + + HadoopConf hadoopConf = build(options); + + assertEquals( + "/keys/gcs.json", + hadoopConf.getExtraOptions().get(GcsHadoopConf.GCS_SERVICE_ACCOUNT_KEY_FILE)); + } + + @Test + void shouldApplyAdditionalPropertiesAndPreferDedicatedKeyFileOption() { + Map<String, String> properties = new HashMap<>(); + properties.put("fs.gs.project.id", "test-project"); + properties.put(GcsHadoopConf.GCS_SERVICE_ACCOUNT_KEY_FILE, "/keys/from-map.json"); + + Map<String, Object> options = new HashMap<>(); + options.put(GcsFileBaseOptions.GCS_PROPERTIES.key(), properties); + options.put(GcsFileBaseOptions.SERVICE_ACCOUNT_KEY_FILE.key(), "/keys/explicit.json"); + + HadoopConf hadoopConf = build(options); + + assertEquals("test-project", hadoopConf.getExtraOptions().get("fs.gs.project.id")); + assertEquals( + "/keys/explicit.json", + hadoopConf.getExtraOptions().get(GcsHadoopConf.GCS_SERVICE_ACCOUNT_KEY_FILE)); + } + + @Test + void shouldRejectBlankServiceAccountKeyFile() { + Map<String, Object> options = new HashMap<>(); + options.put(GcsFileBaseOptions.SERVICE_ACCOUNT_KEY_FILE.key(), " "); + + IllegalArgumentException exception = + assertThrows(IllegalArgumentException.class, () -> build(options)); + + assertTrue(exception.getMessage().contains("must not be blank")); + } + + @Test + void shouldRejectInvalidBucketUris() { + String[] invalidBuckets = { + "test-bucket", + "s3a://test-bucket", + "gs://test-bucket/path", + "gs://user@test-bucket", + "gs://test-bucket?query=value", + "gs://test-bucket#fragment", + "gs://" + }; + + for (String invalidBucket : invalidBuckets) { + Map<String, Object> options = new HashMap<>(); + options.put(GcsFileBaseOptions.BUCKET.key(), invalidBucket); + + assertThrows( + IllegalArgumentException.class, + () -> GcsHadoopConf.buildWithReadonlyConfig(ReadonlyConfig.fromMap(options)), + invalidBucket); + } + } + + private static HadoopConf build(Map<String, Object> options) { + Map<String, Object> config = new HashMap<>(options); + config.put(GcsFileBaseOptions.BUCKET.key(), "gs://test-bucket"); + return GcsHadoopConf.buildWithReadonlyConfig(ReadonlyConfig.fromMap(config)); + } +} diff --git a/seatunnel-connectors-v2/connector-file/pom.xml b/seatunnel-connectors-v2/connector-file/pom.xml index 44ee201599..bed8eaa1f2 100644 --- a/seatunnel-connectors-v2/connector-file/pom.xml +++ b/seatunnel-connectors-v2/connector-file/pom.xml @@ -38,6 +38,7 @@ <module>connector-file-base-hadoop</module> <module>connector-file-sftp</module> <module>connector-file-s3</module> + <module>connector-file-gcs</module> <module>connector-file-obs</module> <module>connector-file-jindo-oss</module> <module>connector-file-cos</module> diff --git a/seatunnel-dist/pom.xml b/seatunnel-dist/pom.xml index ae00d170db..15156d53f8 100644 --- a/seatunnel-dist/pom.xml +++ b/seatunnel-dist/pom.xml @@ -596,6 +596,12 @@ <version>${project.version}</version> <scope>provided</scope> </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-file-gcs</artifactId> + <version>${project.version}</version> + <scope>provided</scope> + </dependency> <dependency> <groupId>org.apache.seatunnel</groupId> <artifactId>connector-amazondynamodb</artifactId> diff --git a/seatunnel-dist/release-docs/NOTICE b/seatunnel-dist/release-docs/NOTICE index fd355b4d9a..c9ebe8f4bb 100644 --- a/seatunnel-dist/release-docs/NOTICE +++ b/seatunnel-dist/release-docs/NOTICE @@ -874,6 +874,110 @@ The initial codebase was donated to the ASF by HugeGraph Authors, copyright 2017 + Copyright 2019-2024 Couchbase, Inc. + Licensed under the Apache License, Version 2.0 +========================================================================= + + +=============================================================================== + + Third-party dependencies for GCS file connector: + +=============================================================================== + + com.google.cloud.bigdataoss:gcs-connector (https://github.com/GoogleCloudDataproc/hadoop-connectors) + + Copyright 2013 Google Inc. + + Licensed under the Apache License, Version 2.0 + +The shaded GCS connector includes the following notices: + +src/test/org/apache/commons/codec/language/DoubleMetaphoneTest.java +contains test data from http://aspell.net/test/orig/batch0.tab. +Copyright (C) 2002 Kevin Atkinson ([email protected]) + +This product includes software developed at +The Apache Software Foundation (http://www.apache.org/). + +Apache HttpCore +Copyright 2005-2021 The Apache Software Foundation + +# Jackson JSON processor + +Jackson is a high-performance, Free/Open Source JSON processing library. +It was originally written by Tatu Saloranta ([email protected]), and has +been in development since 2007. +It is currently developed by a community of developers. + +## Licensing + +Jackson 2.x core and extension components are licensed under Apache License 2.0 +To find the details that apply to this artifact see the accompanying LICENSE file. + +## Credits + +A list of contributors may be found from CREDITS(-2.x) file, which is included +in some artifacts (usually source distributions); but is always available +from the source code management (SCM) system project uses. + +Apache HttpClient +Copyright 1999-2020 The Apache Software Foundation + + The Netty Project + ================= + +Please visit the Netty web site for more information: + + * http://netty.io/ + +Copyright 2016 The Netty Project + +The Netty Project 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. + +------------------------------------------------------------------------------- +This product contains a forked and modified version of Tomcat Native + + * LICENSE: + * license/LICENSE.tomcat-native.txt (Apache License 2.0) + * HOMEPAGE: + * http://tomcat.apache.org/native-doc/ + * https://svn.apache.org/repos/asf/tomcat/native/ + +This product contains the Maven wrapper scripts from 'Maven Wrapper', that provides an easy way to ensure a user has everything necessary to run the Maven build. + + * LICENSE: + * license/LICENSE.mvn-wrapper.txt (Apache License 2.0) + * HOMEPAGE: + * https://github.com/takari/maven-wrapper + +This product contains small piece of code to support AIX, taken from netbsd. + + * LICENSE: + * license/LICENSE.aix-netbsd.txt (OpenSSL License) + * HOMEPAGE: + * https://ftp.netbsd.org/pub/NetBSD/NetBSD-current/src/crypto/external/bsd/openssl/dist + +This product contains code from boringssl. + + * LICENSE (Combination ISC and OpenSSL license) + * license/LICENSE.boringssl.txt (Combination ISC and OpenSSL license) + * HOMEPAGE: + * https://boringssl.googlesource.com/boringssl/ + +Apache Commons Codec +Copyright 2002-2017 The Apache Software Foundation + +=============================================================================== + +The content of package org.apache.commons.codec.language.bm has been translated +from the original php source code available at http://stevemorse.org/phoneticinfo.htm +with permission from the original authors. +Original source copyright: +Copyright (c) 2008 Alexander Beider & Stephen P. Morse. + ========================================================================= +=============================================================================== diff --git a/seatunnel-dist/release-docs/licenses/LICENSE-gcs-connector.txt b/seatunnel-dist/release-docs/licenses/LICENSE-gcs-connector.txt new file mode 100644 index 0000000000..c33ad4945b --- /dev/null +++ b/seatunnel-dist/release-docs/licenses/LICENSE-gcs-connector.txt @@ -0,0 +1 @@ +License: {Name: The Apache Software License, Version 2.0, URL: https://www.apache.org/licenses/LICENSE-2.0.txt, Distribution: repo, Comments: Google Cloud Storage Connector for Hadoop (com.google.cloud.bigdataoss:gcs-connector) } diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-gcs-e2e/pom.xml b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-gcs-e2e/pom.xml new file mode 100644 index 0000000000..2413e795ab --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-gcs-e2e/pom.xml @@ -0,0 +1,58 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + 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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.seatunnel</groupId> + <artifactId>seatunnel-connector-v2-e2e</artifactId> + <version>${revision}</version> + </parent> + + <artifactId>connector-file-gcs-e2e</artifactId> + <name>SeaTunnel : E2E : Connector V2 : File GCS</name> + + <dependencies> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-file-gcs</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-assert</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>seatunnel-shade-hadoop3-uber</artifactId> + <scope>test</scope> + <exclusions> + <exclusion> + <groupId>org.apache.avro</groupId> + <artifactId>avro</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>org.testcontainers</groupId> + <artifactId>testcontainers</artifactId> + <scope>test</scope> + </dependency> + </dependencies> +</project> diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-gcs-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/gcs/GcsFileIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-gcs-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/gcs/GcsFileIT.java new file mode 100644 index 0000000000..7bcfb40c1f --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-gcs-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/gcs/GcsFileIT.java @@ -0,0 +1,104 @@ +/* + * 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.gcs; + +import org.apache.seatunnel.e2e.common.TestResource; +import org.apache.seatunnel.e2e.common.TestSuiteBase; +import org.apache.seatunnel.e2e.common.container.EngineType; +import org.apache.seatunnel.e2e.common.container.TestContainer; +import org.apache.seatunnel.e2e.common.container.TestContainerId; +import org.apache.seatunnel.e2e.common.container.seatunnel.SeaTunnelContainer; +import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.TestTemplate; +import org.testcontainers.containers.Container; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.output.Slf4jLogConsumer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.lifecycle.Startables; +import org.testcontainers.utility.DockerImageName; +import org.testcontainers.utility.DockerLoggerFactory; +import org.testcontainers.utility.MountableFile; + +import java.util.stream.Stream; + +@DisabledOnContainer( + value = {TestContainerId.SPARK_2_4}, + type = {EngineType.FLINK}, + disabledReason = "The GCS connector requires Hadoop 3, but these images use Hadoop 2.7") +public class GcsFileIT extends TestSuiteBase implements TestResource { + + private static final String FAKE_GCS_IMAGE = "fsouza/fake-gcs-server:1.56.1"; + private static final String FAKE_GCS_HOST = "fake-gcs"; + private static final int FAKE_GCS_PORT = 4443; + private static final String JOB_CONFIG = "/gcs/gcs_file_to_assert.conf"; + + private GenericContainer<?> fakeGcs; + + @BeforeAll + @Override + public void startUp() { + DockerImageName image = DockerImageName.parse(FAKE_GCS_IMAGE); + fakeGcs = + new GenericContainer<>(image) + .withCommand( + "-scheme", + "http", + "-port", + String.valueOf(FAKE_GCS_PORT), + "-external-url", + "http://"; + FAKE_GCS_HOST + ":" + FAKE_GCS_PORT) + .withNetwork(NETWORK) + .withNetworkAliases(FAKE_GCS_HOST) + .withExposedPorts(FAKE_GCS_PORT) + .withCopyFileToContainer( + MountableFile.forClasspathResource("gcs/data/e2e.json"), + "/data/seatunnel-gcs-test/input/e2e.json") + .withLogConsumer( + new Slf4jLogConsumer( + DockerLoggerFactory.getLogger( + image.asCanonicalNameString()))) + .waitingFor( + Wait.forHttp("/storage/v1/b") + .forPort(FAKE_GCS_PORT) + .forStatusCode(200)); + Startables.deepStart(Stream.of(fakeGcs)).join(); + SeaTunnelContainer.enableGcsOpenCensusThreadExemption(); + } + + @AfterAll + @Override + public void tearDown() { + try { + if (fakeGcs != null) { + fakeGcs.close(); + } + } finally { + SeaTunnelContainer.disableGcsOpenCensusThreadExemption(); + } + } + + @TestTemplate + public void testReadJsonFromGcs(TestContainer container) throws Exception { + Container.ExecResult result = container.executeJob(JOB_CONFIG); + Assertions.assertEquals(0, result.getExitCode(), result.getStderr()); + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-gcs-e2e/src/test/resources/gcs/data/e2e.json b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-gcs-e2e/src/test/resources/gcs/data/e2e.json new file mode 100644 index 0000000000..30c388d30d --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-gcs-e2e/src/test/resources/gcs/data/e2e.json @@ -0,0 +1,2 @@ +{"id":1,"name":"alice"} +{"id":2,"name":"bob"} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-gcs-e2e/src/test/resources/gcs/gcs_file_to_assert.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-gcs-e2e/src/test/resources/gcs/gcs_file_to_assert.conf new file mode 100644 index 0000000000..f5ce0cf5a3 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-gcs-e2e/src/test/resources/gcs/gcs_file_to_assert.conf @@ -0,0 +1,84 @@ +# +# 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. +# + +env { + parallelism = 1 + job.mode = "BATCH" + + spark.app.name = "SeaTunnel" + spark.executor.instances = 1 + spark.executor.cores = 1 + spark.executor.memory = "1g" + spark.master = local +} + +source { + GcsFile { + bucket = "gs://seatunnel-gcs-test" + path = "/input" + file_format_type = "json" + hadoop_gcs_properties = { + "fs.gs.storage.root.url" = "http://fake-gcs:4443/"; + "fs.gs.auth.service.account.enable" = "false" + "fs.gs.auth.null.enable" = "true" + "fs.gs.project.id" = "seatunnel-test" + } + schema = { + fields { + id = int + name = string + } + } + } +} + +sink { + Assert { + rules { + row_rules = [ + { + rule_type = MIN_ROW + rule_value = 2 + }, + { + rule_type = MAX_ROW + rule_value = 2 + } + ] + field_rules = [ + { + field_name = id + field_type = int + field_value = [ + { + rule_type = NOT_NULL + } + ] + }, + { + field_name = name + field_type = string + field_value = [ + { + rule_type = NOT_NULL + } + ] + } + ] + } + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml b/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml index 8b09ffffed..ec50d4656b 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml @@ -46,6 +46,7 @@ <module>connector-file-sftp-e2e</module> <module>connector-file-oss-e2e</module> <module>connector-file-s3-e2e</module> + <module>connector-file-gcs-e2e</module> <module>connector-cassandra-e2e</module> <module>connector-neo4j-e2e</module> <module>connector-http-e2e</module> diff --git a/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/container/seatunnel/SeaTunnelContainer.java b/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/container/seatunnel/SeaTunnelContainer.java index 3764daebff..3fb09b187e 100644 --- a/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/container/seatunnel/SeaTunnelContainer.java +++ b/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/container/seatunnel/SeaTunnelContainer.java @@ -537,6 +537,16 @@ public class SeaTunnelContainer extends AbstractTestContainer { azureQueueE2eActive = false; } + /** Enables GCS OpenCensus thread exemptions while the GCS file E2E test is active. */ + public static void enableGcsOpenCensusThreadExemption() { + gcsE2eActive = true; + } + + /** Disables GCS OpenCensus thread exemptions after the GCS file E2E test completes. */ + public static void disableGcsOpenCensusThreadExemption() { + gcsE2eActive = false; + } + /** * {@code true} while the Couchbase E2E test ({@code CouchbaseIT}) is active. * @@ -551,6 +561,9 @@ public class SeaTunnelContainer extends AbstractTestContainer { /** {@code true} while the Azure Queue Storage E2E test is active. */ static volatile boolean azureQueueE2eActive = false; + /** {@code true} while the GCS file E2E test is active. */ + static volatile boolean gcsE2eActive = false; + /** The thread should be recycled but not, we should fix it in the future. */ protected boolean isIssueWeAlreadyKnow(String threadName) { // Couchbase SDK JVM-global static singleton threads. @@ -588,6 +601,11 @@ public class SeaTunnelContainer extends AbstractTestContainer { if (isAzureQueueReactorThreadExempt(threadName)) { return true; } + // The shaded GCS client's OpenCensus exporters are JVM-global daemon threads. Their names + // are shared by other OpenCensus users, so exempt them only for the GCS E2E lifecycle. + if (isGcsOpenCensusThreadExempt(threadName)) { + return true; + } // ClickHouse com.clickhouse.client.ClickHouseClientBuilder return threadName.startsWith("ClickHouseClientWorker") // InfluxDB okio.AsyncTimeout$Watchdog @@ -631,6 +649,12 @@ public class SeaTunnelContainer extends AbstractTestContainer { || (threadName.startsWith("boundedElastic-evictor-") && azureQueueE2eActive); } + static boolean isGcsOpenCensusThreadExempt(String threadName) { + return gcsE2eActive + && (threadName.startsWith("ExportComponent.ServiceExporterThread-") + || threadName.startsWith("OpenCensus.Disruptor-")); + } + @Override public Container.ExecResult savepointJob(String jobId) throws IOException, InterruptedException { diff --git a/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/container/seatunnel/SeaTunnelContainerThreadExemptionTest.java b/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/container/seatunnel/SeaTunnelContainerThreadExemptionTest.java index 8f382da40c..63bbf34ac3 100644 --- a/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/container/seatunnel/SeaTunnelContainerThreadExemptionTest.java +++ b/seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/container/seatunnel/SeaTunnelContainerThreadExemptionTest.java @@ -30,7 +30,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; /** - * Regression tests for the Couchbase SDK thread-exemption routing fix (Blocker 2). + * Regression tests for lifecycle-scoped third-party client thread exemptions. * * <p>Before the fix, four Couchbase-specific thread names were exempted inside the <em>global</em> * {@code isSystemThread()} method. This meant any Reactor-based connector that leaked a scheduler @@ -65,12 +65,13 @@ class SeaTunnelContainerThreadExemptionTest { private static SeaTunnelContainer containerMock; /** - * Resets the Couchbase lifecycle flag after every test that may have set it, so that tests - * remain isolated regardless of execution order. + * Resets lifecycle flags after every test that may have set one, so tests remain isolated + * regardless of execution order. */ @AfterEach - void resetCouchbaseFlag() { + void resetLifecycleFlags() { SeaTunnelContainer.disableCouchbaseParallelThreadExemption(); + SeaTunnelContainer.disableGcsOpenCensusThreadExemption(); } @BeforeAll @@ -219,6 +220,21 @@ class SeaTunnelContainerThreadExemptionTest { "Thread 'parallel-reactor' must not match the Couchbase numeric-only pattern"); } + @ParameterizedTest(name = "GCS OpenCensus thread {0} is scoped to the GCS E2E lifecycle") + @ValueSource(strings = {"ExportComponent.ServiceExporterThread-0", "OpenCensus.Disruptor-0"}) + void isIssueWeAlreadyKnow_gcsOpenCensusThreads_areLifecycleScoped(String threadName) + throws Exception { + assertFalse( + isIssueWeAlreadyKnow(threadName), + "GCS OpenCensus threads must not be exempted outside the GCS E2E lifecycle"); + + SeaTunnelContainer.enableGcsOpenCensusThreadExemption(); + + assertTrue( + isIssueWeAlreadyKnow(threadName), + "GCS OpenCensus threads must be exempted while the GCS E2E is active"); + } + // ------------------------------------------------------------------------- // Sanity — existing global system threads must still pass isSystemThread // -------------------------------------------------------------------------
