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-12551-53733a112c9e935c5d97f7f24be249caa43b4fd5 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 4c4fd615d677695b619ca0e04fcfd82e104cd4a9 Author: Pravin Jaywantrao Maske <[email protected]> AuthorDate: Wed Sep 30 19:38:47 2026 +0000 [Feature][Zeta] Support Google Cloud Storage for checkpoint and IMap storage (#12551) --- docs/en/engines/zeta/checkpoint-storage.md | 49 ++++++++++++- docs/en/engines/zeta/hybrid-cluster-deployment.md | 27 +++++++ .../engines/zeta/separated-cluster-deployment.md | 28 ++++++++ docs/zh/engines/zeta/checkpoint-storage.md | 49 ++++++++++++- docs/zh/engines/zeta/hybrid-cluster-deployment.md | 28 ++++++++ .../engines/zeta/separated-cluster-deployment.md | 28 ++++++++ .../storage/hdfs/HdfsStorageFactory.java | 10 ++- .../storage/hdfs/common/FileConfiguration.java | 3 +- .../storage/hdfs/common/GcsConfiguration.java | 66 +++++++++++++++++ .../storage/hdfs/common/GcsConfigurationTest.java | 83 ++++++++++++++++++++++ .../storage/file/config/FileConfiguration.java | 3 +- .../imap/storage/file/config/GcsConfiguration.java | 62 ++++++++++++++++ .../storage/file/wal/DiscoveryWalFileFactory.java | 4 ++ .../imap/storage/file/wal/writer/GcsWriter.java | 26 +++++++ .../storage/file/config/GcsConfigurationTest.java | 83 ++++++++++++++++++++++ 15 files changed, 544 insertions(+), 5 deletions(-) diff --git a/docs/en/engines/zeta/checkpoint-storage.md b/docs/en/engines/zeta/checkpoint-storage.md index 9d47be3cec..f20db8a6b6 100644 --- a/docs/en/engines/zeta/checkpoint-storage.md +++ b/docs/en/engines/zeta/checkpoint-storage.md @@ -14,7 +14,7 @@ Checkpoint Storage is a storage mechanism for storing checkpoint data. SeaTunnel Engine supports the following checkpoint storage types: -- HDFS (OSS,COS,S3,HDFS,LocalFile) +- HDFS (OSS,COS,S3,GCS,HDFS,LocalFile) - LocalFile (native), (it's deprecated: use Hdfs(LocalFile) instead. We use the microkernel design pattern to separate the checkpoint storage module from the engine. This allows users to implement their own checkpoint storage modules. @@ -111,6 +111,53 @@ Please add the following jar to the lib directory: - [cos_api-bundle-5.6.69.jar](https://mvnrepository.com/artifact/com.qcloud/cos_api-bundle/5.6.69) - [hadoop-shaded-guava-1.1.1.jar](https://mvnrepository.com/artifact/org.apache.hadoop.thirdparty/hadoop-shaded-guava/1.1.1) +#### GCS + +Google Cloud Storage based hdfs-file uses the [Hadoop GCS connector](https://github.com/GoogleCloudDataproc/hadoop-connectors/tree/master/gcs). `gcs.bucket` is required and must be a bucket URI such as `gs://your-bucket`. All `fs.gs.*` keys are passed to the connector unchanged. + +If you use a service account key file, you can config like this: + +```yaml +seatunnel: + engine: + checkpoint: + interval: 6000 + timeout: 7000 + storage: + type: hdfs + max-retained: 3 + plugin-config: + namespace: # checkpoint storage parent path, the default value is /seatunnel/checkpoint/ + storage.type: gcs + gcs.bucket: gs://your-bucket + fs.gs.project.id: your-project-id + fs.gs.auth.service.account.json.keyfile: /path/to/service-account-key.json +``` + +If no key file is configured, the connector uses Application Default Credentials, so on GKE with Workload Identity (or on a GCE VM with an attached service account) only the bucket is needed: + +```yaml +seatunnel: + engine: + checkpoint: + interval: 6000 + timeout: 7000 + storage: + type: hdfs + max-retained: 3 + plugin-config: + namespace: # checkpoint storage parent path, the default value is /seatunnel/checkpoint/ + storage.type: gcs + gcs.bucket: gs://your-bucket +``` + +The identity needs read, write and delete permission on the bucket objects, for example the `roles/storage.objectAdmin` role on the bucket. + +Please add the following jar to the lib directory: +- [gcs-connector-hadoop3-2.2.33-shaded.jar](https://mvnrepository.com/artifact/com.google.cloud.bigdataoss/gcs-connector/hadoop3-2.2.33) + +Note: renames use the GCS object move API by default. Services that do not implement it, such as the `fake-gcs-server` emulator, need `fs.gs.operation.move.enable: false`, which makes the connector fall back to copy and delete. + #### S3 S3 based hdfs-file you can refer [hadoop s3 docs](https://hadoop.apache.org/docs/stable/hadoop-aws/tools/hadoop-aws/index.html) to config s3. diff --git a/docs/en/engines/zeta/hybrid-cluster-deployment.md b/docs/en/engines/zeta/hybrid-cluster-deployment.md index bf53fba998..cd81845ca4 100644 --- a/docs/en/engines/zeta/hybrid-cluster-deployment.md +++ b/docs/en/engines/zeta/hybrid-cluster-deployment.md @@ -407,6 +407,33 @@ seatunnel-shade-hadoop3-uber-${seatunnel.shade.hadoop.version}-${seatunnel.shade seatunnel-shade-hadoop-aws-${seatunnel.shade.hadoop-aws.version}-${seatunnel.shade.version}.jar ``` +If you use GCS, you can configure it like this: + +```yaml +map: + engine*: + map-store: + enabled: true + initial-mode: EAGER + factory-class-name: org.apache.seatunnel.engine.server.persistence.FileMapStoreFactory + properties: + type: hdfs + namespace: /seatunnel/imap + clusterName: seatunnel-cluster + storage.type: gcs + gcs.bucket: gs://your-bucket + # optional, Application Default Credentials (e.g. GKE Workload Identity) are used when absent + fs.gs.auth.service.account.json.keyfile: /path/to/service-account-key.json +``` + +Notice: The IMap WAL writer rewrites its current object on every update (the same as for S3 and OSS), and GCS allows about one write per second to the same object name. GCS IMap persistence is therefore best suited to low or moderate IMap update rates. + +Notice: When using GCS, make sure that the following jars are in the lib directory. + +``` +gcs-connector-hadoop3-2.2.33-shaded.jar +seatunnel-shade-hadoop3-uber-${seatunnel.shade.hadoop.version}-${seatunnel.shade.version}.jar +``` ## 6. Configure The SeaTunnel Engine Client diff --git a/docs/en/engines/zeta/separated-cluster-deployment.md b/docs/en/engines/zeta/separated-cluster-deployment.md index e6ca1c0bb7..4a36c03747 100644 --- a/docs/en/engines/zeta/separated-cluster-deployment.md +++ b/docs/en/engines/zeta/separated-cluster-deployment.md @@ -369,6 +369,34 @@ seatunnel-shade-hadoop3-uber-${seatunnel.shade.hadoop.version}-${seatunnel.shade seatunnel-shade-hadoop-aws-${seatunnel.shade.hadoop-aws.version}-${seatunnel.shade.version}.jar ``` +If you use GCS, you can configure it like this: + +```yaml +map: + engine*: + map-store: + enabled: true + initial-mode: EAGER + factory-class-name: org.apache.seatunnel.engine.server.persistence.FileMapStoreFactory + properties: + type: hdfs + namespace: /seatunnel/imap + clusterName: seatunnel-cluster + storage.type: gcs + gcs.bucket: gs://your-bucket + # optional, Application Default Credentials (e.g. GKE Workload Identity) are used when absent + fs.gs.auth.service.account.json.keyfile: /path/to/service-account-key.json +``` + +Notice: The IMap WAL writer rewrites its current object on every update (the same as for S3 and OSS), and GCS allows about one write per second to the same object name. GCS IMap persistence is therefore best suited to low or moderate IMap update rates. + +Notice: When using GCS, make sure that the following jars are in the lib directory. + +``` +gcs-connector-hadoop3-2.2.33-shaded.jar +seatunnel-shade-hadoop3-uber-${seatunnel.shade.hadoop.version}-${seatunnel.shade.version}.jar +``` + ### 4.7 Job Scheduling Strategy When resources are insufficient, the job scheduling strategy can be configured in the following two modes: diff --git a/docs/zh/engines/zeta/checkpoint-storage.md b/docs/zh/engines/zeta/checkpoint-storage.md index a05a304645..2cc3852ec3 100644 --- a/docs/zh/engines/zeta/checkpoint-storage.md +++ b/docs/zh/engines/zeta/checkpoint-storage.md @@ -12,7 +12,7 @@ sidebar_position: 7 SeaTunnel Engine支持以下检查点存储类型: -- HDFS (OSS,COS,S3,HDFS,LocalFile) +- HDFS (OSS,COS,S3,GCS,HDFS,LocalFile) - LocalFile (本地)，(已弃用: 使用HDFS(LocalFile)替代). 我们使用微内核设计模式将检查点存储模块从引擎中分离出来。这允许用户实现他们自己的检查点存储模块。 @@ -109,6 +109,53 @@ seatunnel: - [cos_api-bundle-5.6.69.jar](https://mvnrepository.com/artifact/com.qcloud/cos_api-bundle/5.6.69) - [hadoop-shaded-guava-1.1.1.jar](https://mvnrepository.com/artifact/org.apache.hadoop.thirdparty/hadoop-shaded-guava/1.1.1) +#### GCS + +Google Cloud Storage基于hdfs-file，使用[Hadoop GCS connector](https://github.com/GoogleCloudDataproc/hadoop-connectors/tree/master/gcs)。`gcs.bucket`为必填项，且必须是`gs://your-bucket`形式的bucket URI。所有`fs.gs.*`配置项都会原样传递给connector。 + +如果您使用服务账号密钥文件，可以这样配置: + +```yaml +seatunnel: + engine: + checkpoint: + interval: 6000 + timeout: 7000 + storage: + type: hdfs + max-retained: 3 + plugin-config: + namespace: # 检查点存储父路径，默认值为/seatunnel/checkpoint/ + storage.type: gcs + gcs.bucket: gs://your-bucket + fs.gs.project.id: your-project-id + fs.gs.auth.service.account.json.keyfile: /path/to/service-account-key.json +``` + +如果未配置密钥文件，connector会使用Application Default Credentials，因此在启用Workload Identity的GKE上（或绑定了服务账号的GCE虚拟机上）只需要配置bucket: + +```yaml +seatunnel: + engine: + checkpoint: + interval: 6000 + timeout: 7000 + storage: + type: hdfs + max-retained: 3 + plugin-config: + namespace: # 检查点存储父路径，默认值为/seatunnel/checkpoint/ + storage.type: gcs + gcs.bucket: gs://your-bucket +``` + +所使用的身份需要对bucket中的对象具有读、写和删除权限，例如在bucket上授予`roles/storage.objectAdmin`角色。 + +使用前请将如下jar添加到lib目录下： +- [gcs-connector-hadoop3-2.2.33-shaded.jar](https://mvnrepository.com/artifact/com.google.cloud.bigdataoss/gcs-connector/hadoop3-2.2.33) + +注意：默认情况下重命名操作使用GCS对象move API。对于未实现该API的服务（例如`fake-gcs-server`模拟器），需要配置`fs.gs.operation.move.enable: false`，connector会改为先复制再删除。 + #### S3 S3基于hdfs-file，所以你可以参考[Hadoop s3文档](https://hadoop.apache.org/docs/stable/hadoop-aws/tools/hadoop-aws/index.html)来配置s3。 diff --git a/docs/zh/engines/zeta/hybrid-cluster-deployment.md b/docs/zh/engines/zeta/hybrid-cluster-deployment.md index f17b1d2483..142fd79823 100644 --- a/docs/zh/engines/zeta/hybrid-cluster-deployment.md +++ b/docs/zh/engines/zeta/hybrid-cluster-deployment.md @@ -362,6 +362,34 @@ netty-common-4.1.89.Final.jar seatunnel-shade-hadoop3-uber-${seatunnel.shade.hadoop.version}-${seatunnel.shade.version}.jar ``` +如果您使用 GCS，可以像这样配置： + +```yaml +map: + engine*: + map-store: + enabled: true + initial-mode: EAGER + factory-class-name: org.apache.seatunnel.engine.server.persistence.FileMapStoreFactory + properties: + type: hdfs + namespace: /seatunnel/imap + clusterName: seatunnel-cluster + storage.type: gcs + gcs.bucket: gs://your-bucket + # 可选，未配置时使用 Application Default Credentials（例如 GKE Workload Identity） + fs.gs.auth.service.account.json.keyfile: /path/to/service-account-key.json +``` + +注意：IMap 的 WAL 写入器每次更新都会重写当前对象（与 S3、OSS 相同），而 GCS 对同一对象名的写入频率限制约为每秒一次，因此 GCS 作为 IMap 持久化存储更适合 IMap 更新频率较低或中等的场景。 + +注意：使用 GCS 时，确保 lib 目录下有这几个jar。 + +``` +gcs-connector-hadoop3-2.2.33-shaded.jar +seatunnel-shade-hadoop3-uber-${seatunnel.shade.hadoop.version}-${seatunnel.shade.version}.jar +``` + ## 6. 配置 SeaTunnel Engine 客户端 所有 SeaTunnel Engine 客户端的配置都在 `hazelcast-client.yaml` 里。 diff --git a/docs/zh/engines/zeta/separated-cluster-deployment.md b/docs/zh/engines/zeta/separated-cluster-deployment.md index 7796bd2b72..6fce1b6868 100644 --- a/docs/zh/engines/zeta/separated-cluster-deployment.md +++ b/docs/zh/engines/zeta/separated-cluster-deployment.md @@ -334,6 +334,34 @@ netty-common-4.1.89.Final.jar seatunnel-shade-hadoop3-uber-${seatunnel.shade.hadoop.version}-${seatunnel.shade.version}.jar ``` +如果您使用 GCS，可以像这样配置： + +```yaml +map: + engine*: + map-store: + enabled: true + initial-mode: EAGER + factory-class-name: org.apache.seatunnel.engine.server.persistence.FileMapStoreFactory + properties: + type: hdfs + namespace: /seatunnel/imap + clusterName: seatunnel-cluster + storage.type: gcs + gcs.bucket: gs://your-bucket + # 可选，未配置时使用 Application Default Credentials（例如 GKE Workload Identity） + fs.gs.auth.service.account.json.keyfile: /path/to/service-account-key.json +``` + +注意：IMap 的 WAL 写入器每次更新都会重写当前对象（与 S3、OSS 相同），而 GCS 对同一对象名的写入频率限制约为每秒一次，因此 GCS 作为 IMap 持久化存储更适合 IMap 更新频率较低或中等的场景。 + +注意：使用 GCS 时，确保 lib 目录下有这几个jar。 + +``` +gcs-connector-hadoop3-2.2.33-shaded.jar +seatunnel-shade-hadoop3-uber-${seatunnel.shade.hadoop.version}-${seatunnel.shade.version}.jar +``` + ### 4.7 作业调度策略 当资源不足时，作业调度策略可以配置为以下两种模式： diff --git a/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/main/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/HdfsStorageFactory.java b/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/main/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/HdfsStorageFactory.java index eec10aeabe..a3f3e932aa 100644 --- a/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/main/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/HdfsStorageFactory.java +++ b/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/main/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/HdfsStorageFactory.java @@ -34,7 +34,7 @@ import java.util.Map; * following configuration in the configuration file: * * <pre> - * storage.type = hdfs # hdfs, local(default),s3, oss + * storage.type = hdfs # hdfs, local(default), s3, oss, cos, gcs * </pre> * * then you need to configure the following parameters by the storage.type: hdfs {@link @@ -59,6 +59,14 @@ import java.util.Map; * fs.oss.endpoint = "such as: oss-cn-hangzhou.aliyuncs.com" * oss.bucket= "oss://your bucket" * </pre> + * + * gcs {@link org.apache.seatunnel.engine.checkpoint.storage.hdfs.common.GcsConfiguration} eg: gcs + * + * <pre> + * storage.type = "gcs" + * gcs.bucket = "gs://your bucket" + * fs.gs.auth.service.account.json.keyfile = "/path/to/key.json" # optional, ADC if absent + * </pre> */ @AutoService(CheckpointStorageFactory.class) public class HdfsStorageFactory implements CheckpointStorageFactory { diff --git a/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/main/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/common/FileConfiguration.java b/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/main/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/common/FileConfiguration.java index b313cc3905..b6d7597d32 100644 --- a/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/main/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/common/FileConfiguration.java +++ b/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/main/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/common/FileConfiguration.java @@ -25,7 +25,8 @@ public enum FileConfiguration { HDFS("hdfs", new HdfsConfiguration()), S3("s3", new S3Configuration()), OSS("oss", new OssConfiguration()), - COS("cos", new CosConfiguration()); + COS("cos", new CosConfiguration()), + GCS("gcs", new GcsConfiguration()); /** file system type */ private final String name; diff --git a/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/main/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/common/GcsConfiguration.java b/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/main/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/common/GcsConfiguration.java new file mode 100644 index 0000000000..44c43dd05c --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/main/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/common/GcsConfiguration.java @@ -0,0 +1,66 @@ +/* + * 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.engine.checkpoint.storage.hdfs.common; + +import org.apache.hadoop.conf.Configuration; + +import java.util.Map; + +import static org.apache.hadoop.fs.FileSystem.FS_DEFAULT_NAME_KEY; + +/** + * Google Cloud Storage configuration for checkpoint storage, backed by the Hadoop GCS connector + * ({@code gcs-connector}), which must be placed in the lib directory. + * + * <p>All {@code fs.gs.*} keys are passed through to the connector, e.g. {@code + * fs.gs.auth.service.account.json.keyfile}. Without a key file the connector falls back to + * Application Default Credentials, which covers GKE Workload Identity. + */ +public class GcsConfiguration extends AbstractConfiguration { + + /** ************** GCS required keys ************** */ + public static final String GCS_BUCKET_KEY = "gcs.bucket"; + + /* GCS constants */ + private static final String GCS_SCHEME_PREFIX = "gs://"; + private static final String GCS_IMPL_KEY = "fs.gs.impl"; + private static final String HDFS_GCS_IMPL = + "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"; + private static final String GCS_KEY = "fs.gs."; + + @Override + public Configuration buildConfiguration(Map<String, String> config) { + checkConfiguration(config, GCS_BUCKET_KEY); + String bucket = config.get(GCS_BUCKET_KEY); + if (!bucket.startsWith(GCS_SCHEME_PREFIX)) { + throw new IllegalArgumentException( + String.format( + "%s must be a GCS bucket URI such as 'gs://my-bucket', but was '%s'", + GCS_BUCKET_KEY, bucket)); + } + Configuration hadoopConf = new Configuration(); + hadoopConf.set(FS_DEFAULT_NAME_KEY, bucket); + hadoopConf.set(GCS_IMPL_KEY, HDFS_GCS_IMPL); + hadoopConf.setBoolean( + String.format(COMMON_DISABLE_CACHE, GCS_IMPL_KEY), + Boolean.parseBoolean( + config.getOrDefault(DISABLE_CACHE_KEY, DISABLE_CACHE_DEFAULT_VALUE))); + setExtraConfiguration(hadoopConf, config, GCS_KEY); + return hadoopConf; + } +} diff --git a/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/test/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/common/GcsConfigurationTest.java b/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/test/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/common/GcsConfigurationTest.java new file mode 100644 index 0000000000..d06c625187 --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/test/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/common/GcsConfigurationTest.java @@ -0,0 +1,83 @@ +/* + * 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.engine.checkpoint.storage.hdfs.common; + +import org.apache.hadoop.conf.Configuration; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.Map; + +import static org.apache.hadoop.fs.FileSystem.FS_DEFAULT_NAME_KEY; + +public class GcsConfigurationTest { + + @Test + public void testBuildConfiguration() throws Exception { + Map<String, String> config = new HashMap<>(); + config.put("gcs.bucket", "gs://seatunnel-checkpoint"); + config.put("fs.gs.auth.service.account.json.keyfile", "/path/to/key.json"); + config.put("fs.s3a.access.key", "must-not-leak"); + + Configuration hadoopConf = + FileConfiguration.valueOf("GCS").getConfiguration().buildConfiguration(config); + + Assertions.assertEquals("gs://seatunnel-checkpoint", hadoopConf.get(FS_DEFAULT_NAME_KEY)); + Assertions.assertEquals( + "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem", + hadoopConf.get("fs.gs.impl")); + Assertions.assertTrue(hadoopConf.getBoolean("fs.gs.impl.disable.cache", false)); + Assertions.assertEquals( + "/path/to/key.json", hadoopConf.get("fs.gs.auth.service.account.json.keyfile")); + Assertions.assertNull(hadoopConf.get("fs.s3a.access.key")); + } + + @Test + public void testDisableCacheCanBeTurnedOff() throws Exception { + Map<String, String> config = new HashMap<>(); + config.put("gcs.bucket", "gs://seatunnel-checkpoint"); + config.put("disable.cache", "false"); + + Configuration hadoopConf = new GcsConfiguration().buildConfiguration(config); + + Assertions.assertFalse(hadoopConf.getBoolean("fs.gs.impl.disable.cache", true)); + } + + @Test + public void testBucketIsRequired() { + IllegalArgumentException e = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> new GcsConfiguration().buildConfiguration(new HashMap<>())); + Assertions.assertEquals("gcs.bucket is required", e.getMessage()); + } + + @Test + public void testBucketMustUseGsScheme() { + Map<String, String> config = new HashMap<>(); + config.put("gcs.bucket", "seatunnel-checkpoint"); + + IllegalArgumentException e = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> new GcsConfiguration().buildConfiguration(config)); + Assertions.assertTrue(e.getMessage().contains("gs://my-bucket")); + } +} diff --git a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/config/FileConfiguration.java b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/config/FileConfiguration.java index 2da8a5056f..56db2b19dc 100644 --- a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/config/FileConfiguration.java +++ b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/config/FileConfiguration.java @@ -23,7 +23,8 @@ package org.apache.seatunnel.engine.imap.storage.file.config; public enum FileConfiguration { HDFS("hdfs", new HdfsConfiguration()), S3("s3", new S3Configuration()), - OSS("oss", new OssConfiguration()); + OSS("oss", new OssConfiguration()), + GCS("gcs", new GcsConfiguration()); /** file system type */ private final String name; diff --git a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/config/GcsConfiguration.java b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/config/GcsConfiguration.java new file mode 100644 index 0000000000..557eabd68c --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/config/GcsConfiguration.java @@ -0,0 +1,62 @@ +/* + * 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.engine.imap.storage.file.config; + +import org.apache.hadoop.conf.Configuration; + +import java.util.Map; + +import static org.apache.hadoop.fs.FileSystem.FS_DEFAULT_NAME_KEY; + +/** + * Google Cloud Storage configuration for IMap persistence, backed by the Hadoop GCS connector + * ({@code gcs-connector}), which must be placed in the lib directory. + * + * <p>All {@code fs.gs.*} keys are passed through to the connector. Without {@code + * fs.gs.auth.service.account.json.keyfile} the connector falls back to Application Default + * Credentials, which covers GKE Workload Identity. + */ +public class GcsConfiguration extends AbstractConfiguration { + + public static final String GCS_BUCKET_KEY = "gcs.bucket"; + + private static final String GCS_SCHEME_PREFIX = "gs://"; + private static final String GCS_IMPL_KEY = "fs.gs.impl"; + private static final String HDFS_GCS_IMPL = + "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"; + private static final String GCS_KEY = "fs.gs."; + + @Override + public Configuration buildConfiguration(Map<String, String> config) { + checkConfiguration(config, GCS_BUCKET_KEY); + String bucket = config.get(GCS_BUCKET_KEY); + if (!bucket.startsWith(GCS_SCHEME_PREFIX)) { + throw new IllegalArgumentException( + String.format( + "%s must be a GCS bucket URI such as 'gs://my-bucket', but was '%s'", + GCS_BUCKET_KEY, bucket)); + } + Configuration hadoopConf = new Configuration(); + hadoopConf.set(FS_DEFAULT_NAME_KEY, bucket); + hadoopConf.set(GCS_IMPL_KEY, HDFS_GCS_IMPL); + // Unlike the checkpoint storage GcsConfiguration, the FileSystem cache is intentionally + // left at the Hadoop default, matching the IMap OSS and S3 configurations. + setExtraConfiguration(hadoopConf, config, GCS_KEY); + return hadoopConf; + } +} diff --git a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/wal/DiscoveryWalFileFactory.java b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/wal/DiscoveryWalFileFactory.java index 7f76ae2f39..8545d2b6f3 100644 --- a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/wal/DiscoveryWalFileFactory.java +++ b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/wal/DiscoveryWalFileFactory.java @@ -22,6 +22,7 @@ package org.apache.seatunnel.engine.imap.storage.file.wal; import org.apache.seatunnel.engine.imap.storage.file.config.FileConfiguration; import org.apache.seatunnel.engine.imap.storage.file.wal.reader.DefaultReader; import org.apache.seatunnel.engine.imap.storage.file.wal.reader.IFileReader; +import org.apache.seatunnel.engine.imap.storage.file.wal.writer.GcsWriter; import org.apache.seatunnel.engine.imap.storage.file.wal.writer.HdfsWriter; import org.apache.seatunnel.engine.imap.storage.file.wal.writer.IFileWriter; import org.apache.seatunnel.engine.imap.storage.file.wal.writer.OssWriter; @@ -35,6 +36,7 @@ public class DiscoveryWalFileFactory { case HDFS: case S3: case OSS: + case GCS: return new DefaultReader(); } throw new UnsupportedOperationException("Unsupported type " + type); @@ -49,6 +51,8 @@ public class DiscoveryWalFileFactory { return new S3Writer(); case OSS: return new OssWriter(); + case GCS: + return new GcsWriter(); } throw new UnsupportedOperationException("Unsupported type " + type); } diff --git a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/wal/writer/GcsWriter.java b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/wal/writer/GcsWriter.java new file mode 100644 index 0000000000..db6acdc8c3 --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/wal/writer/GcsWriter.java @@ -0,0 +1,26 @@ +/* + * 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.engine.imap.storage.file.wal.writer; + +/** WAL writer for Google Cloud Storage; GCS has no append, so it reuses {@link CloudWriter}. */ +public class GcsWriter extends CloudWriter { + @Override + public String identifier() { + return "gcs"; + } +} diff --git a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/config/GcsConfigurationTest.java b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/config/GcsConfigurationTest.java new file mode 100644 index 0000000000..2e08615130 --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/config/GcsConfigurationTest.java @@ -0,0 +1,83 @@ +/* + * 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.engine.imap.storage.file.config; + +import org.apache.seatunnel.engine.imap.storage.file.wal.DiscoveryWalFileFactory; +import org.apache.seatunnel.engine.imap.storage.file.wal.reader.DefaultReader; +import org.apache.seatunnel.engine.imap.storage.file.wal.writer.GcsWriter; + +import org.apache.hadoop.conf.Configuration; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.Map; + +import static org.apache.hadoop.fs.FileSystem.FS_DEFAULT_NAME_KEY; + +public class GcsConfigurationTest { + + @Test + public void testBuildConfiguration() throws Exception { + Map<String, String> config = new HashMap<>(); + config.put("gcs.bucket", "gs://seatunnel-imap"); + config.put("fs.gs.auth.service.account.json.keyfile", "/path/to/key.json"); + config.put("block.size", "2097152"); + config.put("fs.oss.accessKeyId", "must-not-leak"); + + GcsConfiguration gcsConfiguration = + (GcsConfiguration) FileConfiguration.valueOf("GCS").getConfiguration(); + Configuration hadoopConf = gcsConfiguration.buildConfiguration(config); + + Assertions.assertEquals("gs://seatunnel-imap", hadoopConf.get(FS_DEFAULT_NAME_KEY)); + Assertions.assertEquals( + "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem", + hadoopConf.get("fs.gs.impl")); + Assertions.assertEquals( + "/path/to/key.json", hadoopConf.get("fs.gs.auth.service.account.json.keyfile")); + Assertions.assertNull(hadoopConf.get("fs.oss.accessKeyId")); + Assertions.assertEquals(2097152L, gcsConfiguration.getBlockSize()); + } + + @Test + public void testBucketIsRequired() { + IllegalArgumentException e = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> new GcsConfiguration().buildConfiguration(new HashMap<>())); + Assertions.assertEquals("gcs.bucket is required", e.getMessage()); + } + + @Test + public void testBucketMustUseGsScheme() { + Map<String, String> config = new HashMap<>(); + config.put("gcs.bucket", "s3a://seatunnel-imap"); + + Assertions.assertThrows( + IllegalArgumentException.class, + () -> new GcsConfiguration().buildConfiguration(config)); + } + + @Test + public void testWalUsesCloudWriterForGcs() { + Assertions.assertTrue(DiscoveryWalFileFactory.getWriter("gcs") instanceof GcsWriter); + Assertions.assertEquals("gcs", DiscoveryWalFileFactory.getWriter("gcs").identifier()); + Assertions.assertTrue(DiscoveryWalFileFactory.getReader("gcs") instanceof DefaultReader); + } +}
