This is an automated email from the ASF dual-hosted git repository.
leekeiabstraction pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 69eebd0fc feat: [FLUSS-2686] Add COS filesystem support (#2836)
69eebd0fc is described below
commit 69eebd0fc456662327465475e7f6e2871043d9ea
Author: ForwardXu <[email protected]>
AuthorDate: Fri May 29 14:23:51 2026 +0800
feat: [FLUSS-2686] Add COS filesystem support (#2836)
* [FLUSS-2686] Add COS filesystem support
This adds a new fluss-fs-cos module that integrates Tencent Cloud COS
(Cloud Object Storage) as a remote filesystem for Fluss.
Key changes:
- Add fluss-fs-cos module with hadoop-cos and cos_api dependencies
- Implement COSFileSystem extending HadoopFileSystem (scheme: cosn)
- Implement COSFileSystemPlugin as FileSystemPlugin SPI
- Add security token support (COSSecurityTokenProvider/Receiver)
- Add DynamicTemporaryCOSCredentialsProvider for temporary credentials
- Add integration tests for COS filesystem behavior
- Register module in fluss-filesystems parent pom
* [FLUSS-2686] Add COS filesystem support
* fix some comments
* fix some comments
* Improve COS filesystem: fix STS token provider, add temporary credentials
doc, and address review comments
* Fix parent POM version and update config
* [fluss-fs-cos] Fix NOTICE file: update outdated dependencies and add
missing ones
* [filesystem] Address review comments for COS filesystem
- Rename Maven property fs.cosn.sdk.version to fs.hadoop.cos.version to
accurately reflect that it is the hadoop-cos version rather than the Tencent
Cloud COS SDK version.
- Align COSSecurityTokenProvider default token duration with OSS/OBS (30min
-> 1h).
- Add COSWithCredentialsProviderFileSystemBehaviorITCase to cover the
CREDENTIALS_PROVIDER branch in COSFileSystemPlugin, reusing hadoop-cos's
EnvironmentVariableCredentialsProvider.
* [cos] Rename COSN_ENDPOINT to COSN_ENDPOINT_SUFFIX for consistency with
config key
* [filesystems] Address COS review: scope STS policy, drop duplicate COS
SDK, fix docs
1. STS tokens scoped to remote.data.dir bucket: COSSecurityTokenProvider
now derives a default policy that limits name/cos:* to the bucket (and prefix)
parsed from fsUri. Users can override via fs.cosn.security.token.policy.
2. Drop duplicate COS SDK: exclude com.qcloud:cos_api-bundle from
hadoop-cos so only the directly declared cos_api remains on the classpath,
eliminating 902 overlapping com.qcloud.cos.* classes during shade. NOTICE
updated accordingly.
3. Docs: use the correct hadoop-cos provider class
EnvironmentVariableCredentialsProvider and the correct env vars COSN_SECRET_ID
/ COSN_SECRET_KEY.
4. Docs: align temporary credential lifetime with the code (1 hour, matches
DEFAULT_DURATION_SECONDS=3600L).
5. Docs: remove trailing whitespace in
website/docs/maintenance/filesystems/cos.md.
---
fluss-filesystems/fluss-fs-cos/pom.xml | 273 +++++++++++++++++++++
.../org/apache/fluss/fs/cos/COSFileSystem.java | 67 +++++
.../apache/fluss/fs/cos/COSFileSystemPlugin.java | 144 +++++++++++
.../fs/cos/token/COSSecurityTokenProvider.java | 193 +++++++++++++++
.../fs/cos/token/COSSecurityTokenReceiver.java | 112 +++++++++
.../DynamicTemporaryCOSCredentialsProvider.java | 69 ++++++
.../src/main/resources/META-INF/NOTICE | 77 ++++++
.../main/resources/META-INF/licenses/LICENSE.jaxb | 135 ++++++++++
.../services/org.apache.fluss.fs.FileSystemPlugin | 17 ++
...org.apache.fluss.fs.token.SecurityTokenReceiver | 17 ++
.../fluss/fs/cos/COSFileSystemBehaviorITCase.java | 67 +++++
.../apache/fluss/fs/cos/COSTestCredentials.java | 119 +++++++++
...redentialsProviderFileSystemBehaviorITCase.java | 68 +++++
.../COSWithTokenFileSystemBehaviorBaseITCase.java | 43 ++++
.../cos/COSWithTokenFileSystemBehaviorITCase.java | 72 ++++++
fluss-filesystems/pom.xml | 1 +
fluss-test-coverage/pom.xml | 1 +
website/docs/maintenance/filesystems/cos.md | 89 +++++++
website/docs/maintenance/filesystems/overview.md | 4 +-
19 files changed, 1567 insertions(+), 1 deletion(-)
diff --git a/fluss-filesystems/fluss-fs-cos/pom.xml
b/fluss-filesystems/fluss-fs-cos/pom.xml
new file mode 100644
index 000000000..7887ee2a1
--- /dev/null
+++ b/fluss-filesystems/fluss-fs-cos/pom.xml
@@ -0,0 +1,273 @@
+<?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.fluss</groupId>
+ <artifactId>fluss-filesystems</artifactId>
+ <version>1.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>fluss-fs-cos</artifactId>
+ <name>Fluss : FileSystems : COS FS</name>
+
+ <properties>
+ <fs.hadoop.cos.version>3.3.5</fs.hadoop.cos.version>
+ <fs.cos.api.version>5.6.139</fs.cos.api.version>
+ <fs.cos.sts.sdk.version>3.1.678</fs.cos.sts.sdk.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-common</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-fs-hadoop-shaded</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-fs-hadoop</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-cos</artifactId>
+ <version>${fs.hadoop.cos.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.qcloud</groupId>
+ <artifactId>cos_api</artifactId>
+ </exclusion>
+ <!--
+ Exclude the bundled COS SDK shipped by hadoop-cos to avoid
having two
+ COS SDK variants on the classpath (cos_api-bundle:5.6.69 and
the directly
+ declared cos_api:${fs.cos.api.version}), which produces
hundreds of
+ overlapping com.qcloud.cos.* classes during shading.
+ -->
+ <exclusion>
+ <groupId>com.qcloud</groupId>
+ <artifactId>cos_api-bundle</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>ch.qos.reload4j</groupId>
+ <artifactId>reload4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-reload4j</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>com.qcloud</groupId>
+ <artifactId>cos_api</artifactId>
+ <version>${fs.cos.api.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>javax.xml.bind</groupId>
+ <artifactId>jaxb-api</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>com.tencentcloudapi</groupId>
+ <artifactId>tencentcloud-sdk-java-sts</artifactId>
+ <version>${fs.cos.sts.sdk.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>javax.xml.bind</groupId>
+ <artifactId>jaxb-api</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <!-- Hadoop requires jaxb-api for javax.xml.bind.JAXBException -->
+ <groupId>javax.xml.bind</groupId>
+ <artifactId>jaxb-api</artifactId>
+ <version>${jaxb.api.version}</version>
+ <!-- packaged as an optional dependency that is only accessible on
Java 11+ -->
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-test-utils</artifactId>
+ </dependency>
+ <!-- for the behavior test suite -->
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-common</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>default-jar</id>
+ <goals>
+ <goal>jar</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>test-jar</id>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <archive>
+ <manifestEntries>
+ <!-- jaxb-api is packaged as an optional
dependency that is only accessible on Java 11 -->
+ <Multi-Release>true</Multi-Release>
+ </manifestEntries>
+ </archive>
+ </configuration>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>copy-javax-jars</id>
+ <phase>process-resources</phase>
+ <goals>
+ <goal>copy</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <artifactItems>
+ <artifactItem>
+ <groupId>javax.xml.bind</groupId>
+ <artifactId>jaxb-api</artifactId>
+ <version>${jaxb.api.version}</version>
+ <type>jar</type>
+ <overWrite>true</overWrite>
+ </artifactItem>
+ </artifactItems>
+
<outputDirectory>${project.build.directory}/temporary</outputDirectory>
+ </configuration>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-antrun-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>unpack-javax-libraries</id>
+ <phase>process-resources</phase>
+ <goals>
+ <goal>run</goal>
+ </goals>
+ <configuration>
+ <target>
+ <echo message="unpacking javax jars"/>
+ <unzip
dest="${project.build.directory}/classes/META-INF/versions/11">
+ <fileset
dir="${project.build.directory}/temporary">
+ <include name="*"/>
+ </fileset>
+ </unzip>
+ </target>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>shade-fluss</id>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <artifactSet>
+ <includes>
+ <include>*:*</include>
+ </includes>
+ <excludes>
+
<exclude>javax.servlet:servlet-api</exclude>
+ <exclude>xmlenc:xmlenc</exclude>
+ </excludes>
+ </artifactSet>
+ <filters>
+ <filter>
+ <artifact>*</artifact>
+ <excludes>
+ <exclude>.gitkeep</exclude>
+ <exclude>mime.types</exclude>
+ <exclude>mozilla/**</exclude>
+ <exclude>LICENSE.txt</exclude>
+ <exclude>license/LICENSE*</exclude>
+
<exclude>okhttp3/internal/publicsuffix/NOTICE</exclude>
+ <exclude>NOTICE</exclude>
+ </excludes>
+ </filter>
+ <filter>
+
<artifact>org.apache.fluss:fluss-fs-hadoop</artifact>
+ <excludes>
+ <exclude>META-INF/**</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <relocations>
+ <relocation>
+ <pattern>org.apache.commons</pattern>
+
<shadedPattern>org.apache.fluss.shaded.org.apache.commons</shadedPattern>
+ </relocation>
+ </relocations>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ </plugins>
+ </build>
+
+</project>
diff --git
a/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/COSFileSystem.java
b/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/COSFileSystem.java
new file mode 100644
index 000000000..acb33e18d
--- /dev/null
+++
b/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/COSFileSystem.java
@@ -0,0 +1,67 @@
+/*
+ * 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.fluss.fs.cos;
+
+import org.apache.fluss.fs.cos.token.COSSecurityTokenProvider;
+import org.apache.fluss.fs.hdfs.HadoopFileSystem;
+import org.apache.fluss.fs.token.ObtainedSecurityToken;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+
+import java.io.IOException;
+import java.net.URI;
+
+/**
+ * A {@link FileSystem} for Tencent Cloud COS that wraps an {@link
HadoopFileSystem}, but overwrite
+ * method to generate access security token.
+ */
+class COSFileSystem extends HadoopFileSystem {
+
+ private final Configuration conf;
+ private volatile COSSecurityTokenProvider cosSecurityTokenProvider;
+ private final String scheme;
+ private final URI fsUri;
+
+ COSFileSystem(FileSystem hadoopFileSystem, String scheme, URI fsUri,
Configuration conf) {
+ super(hadoopFileSystem);
+ this.scheme = scheme;
+ this.fsUri = fsUri;
+ this.conf = conf;
+ }
+
+ @Override
+ public ObtainedSecurityToken obtainSecurityToken() throws IOException {
+ try {
+ mayCreateSecurityTokenProvider();
+ return cosSecurityTokenProvider.obtainSecurityToken(scheme);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ private void mayCreateSecurityTokenProvider() throws IOException {
+ if (cosSecurityTokenProvider == null) {
+ synchronized (this) {
+ if (cosSecurityTokenProvider == null) {
+ cosSecurityTokenProvider = new
COSSecurityTokenProvider(fsUri, conf);
+ }
+ }
+ }
+ }
+}
diff --git
a/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/COSFileSystemPlugin.java
b/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/COSFileSystemPlugin.java
new file mode 100644
index 000000000..95309eee6
--- /dev/null
+++
b/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/COSFileSystemPlugin.java
@@ -0,0 +1,144 @@
+/*
+ * 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.fluss.fs.cos;
+
+import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.config.ConfigBuilder;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.fs.FileSystem;
+import org.apache.fluss.fs.FileSystemPlugin;
+import org.apache.fluss.fs.cos.token.COSSecurityTokenReceiver;
+
+import org.apache.hadoop.fs.cosn.CosNFileSystem;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+
+/** Simple factory for the Tencent Cloud COS file system. */
+public class COSFileSystemPlugin implements FileSystemPlugin {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(COSFileSystemPlugin.class);
+
+ public static final String SCHEME = "cosn";
+
+ /**
+ * In order to simplify, we make fluss cos configuration keys same with
hadoop cos module. So,
+ * we add all configuration key with prefix `fs.cosn` in fluss conf to
hadoop conf.
+ */
+ private static final String[] FLUSS_CONFIG_PREFIXES = {"fs.cosn."};
+
+ public static final String SECRET_ID = "fs.cosn.userinfo.secretId";
+ public static final String SECRET_KEY = "fs.cosn.userinfo.secretKey";
+ public static final String CREDENTIALS_PROVIDER =
"fs.cosn.credentials.provider";
+
+ public static final String REGION = "fs.cosn.userinfo.region";
+
+ public static final String ENDPOINT_KEY = "fs.cosn.bucket.endpoint_suffix";
+
+ /**
+ * Optional user-provided STS access policy (a JSON string) applied when
calling {@code
+ * GetFederationToken}. When unset, Fluss will derive a default policy
that scopes the temporary
+ * credential to the bucket configured in {@code remote.data.dir} so that
the token cannot
+ * access other COS resources.
+ */
+ public static final String SECURITY_TOKEN_POLICY =
"fs.cosn.security.token.policy";
+
+ @Override
+ public String getScheme() {
+ return SCHEME;
+ }
+
+ @Override
+ public FileSystem create(URI fsUri, Configuration flussConfig) throws
IOException {
+ org.apache.hadoop.conf.Configuration hadoopConfig =
getHadoopConfiguration(flussConfig);
+
+ // set credential provider
+ if (hadoopConfig.get(SECRET_ID) == null) {
+ String credentialsProvider =
hadoopConfig.get(CREDENTIALS_PROVIDER);
+ if (credentialsProvider != null) {
+ LOG.info(
+ "{} is not set, but {} is set, using credential
provider {}.",
+ SECRET_ID,
+ CREDENTIALS_PROVIDER,
+ credentialsProvider);
+ } else {
+ // no secretId, no credentialsProvider,
+ // set default credential provider which will get token from
+ // COSSecurityTokenReceiver
+ setDefaultCredentialProvider(hadoopConfig);
+ }
+ } else {
+ LOG.info("{} is set, using provided secret id and secret key.",
SECRET_ID);
+ }
+
+ final String scheme = fsUri.getScheme();
+ final String authority = fsUri.getAuthority();
+
+ if (scheme == null && authority == null) {
+ fsUri =
org.apache.hadoop.fs.FileSystem.getDefaultUri(hadoopConfig);
+ } else if (scheme != null && authority == null) {
+ URI defaultUri =
org.apache.hadoop.fs.FileSystem.getDefaultUri(hadoopConfig);
+ if (scheme.equals(defaultUri.getScheme()) &&
defaultUri.getAuthority() != null) {
+ fsUri = defaultUri;
+ }
+ }
+
+ org.apache.hadoop.fs.FileSystem fileSystem = initFileSystem(fsUri,
hadoopConfig);
+ return new COSFileSystem(fileSystem, getScheme(), fsUri, hadoopConfig);
+ }
+
+ protected org.apache.hadoop.fs.FileSystem initFileSystem(
+ URI fsUri, org.apache.hadoop.conf.Configuration hadoopConfig)
throws IOException {
+ CosNFileSystem fileSystem = new CosNFileSystem();
+ fileSystem.initialize(fsUri, hadoopConfig);
+ return fileSystem;
+ }
+
+ protected void
setDefaultCredentialProvider(org.apache.hadoop.conf.Configuration hadoopConfig)
{
+ // use COSSecurityTokenReceiver to update hadoop config to set
credentialsProvider
+ COSSecurityTokenReceiver.updateHadoopConfig(hadoopConfig);
+ }
+
+ @VisibleForTesting
+ org.apache.hadoop.conf.Configuration getHadoopConfiguration(Configuration
flussConfig) {
+ org.apache.hadoop.conf.Configuration conf = new
org.apache.hadoop.conf.Configuration();
+ if (flussConfig == null) {
+ return conf;
+ }
+
+ // read all configuration with prefix 'FLUSS_CONFIG_PREFIXES'
+ for (String key : flussConfig.keySet()) {
+ for (String prefix : FLUSS_CONFIG_PREFIXES) {
+ if (key.startsWith(prefix)) {
+ String value =
+ flussConfig.getString(
+
ConfigBuilder.key(key).stringType().noDefaultValue(), null);
+ conf.set(key, value);
+
+ LOG.debug(
+ "Adding Fluss config entry for {} as {} to Hadoop
config",
+ key,
+ conf.get(key));
+ }
+ }
+ }
+ return conf;
+ }
+}
diff --git
a/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProvider.java
b/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProvider.java
new file mode 100644
index 000000000..5b4390851
--- /dev/null
+++
b/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProvider.java
@@ -0,0 +1,193 @@
+/*
+ * 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.fluss.fs.cos.token;
+
+import org.apache.fluss.fs.token.Credentials;
+import org.apache.fluss.fs.token.CredentialsJsonSerde;
+import org.apache.fluss.fs.token.ObtainedSecurityToken;
+
+import com.tencentcloudapi.common.Credential;
+import com.tencentcloudapi.common.exception.TencentCloudSDKException;
+import com.tencentcloudapi.sts.v20180813.StsClient;
+import com.tencentcloudapi.sts.v20180813.models.GetFederationTokenRequest;
+import com.tencentcloudapi.sts.v20180813.models.GetFederationTokenResponse;
+import org.apache.hadoop.conf.Configuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.URI;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.fluss.fs.cos.COSFileSystemPlugin.ENDPOINT_KEY;
+import static org.apache.fluss.fs.cos.COSFileSystemPlugin.REGION;
+import static org.apache.fluss.fs.cos.COSFileSystemPlugin.SECRET_ID;
+import static org.apache.fluss.fs.cos.COSFileSystemPlugin.SECRET_KEY;
+import static
org.apache.fluss.fs.cos.COSFileSystemPlugin.SECURITY_TOKEN_POLICY;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * A provider to provide Tencent Cloud COS security token by calling STS
GetFederationToken API to
+ * obtain temporary credentials.
+ */
+public class COSSecurityTokenProvider {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(COSSecurityTokenProvider.class);
+
+ /** Default federation token name. */
+ private static final String FEDERATION_TOKEN_NAME = "fluss-cos-federation";
+
+ /** Default duration seconds for temporary credentials, 3600s = 1h. */
+ private static final long DEFAULT_DURATION_SECONDS = 3600L;
+
+ private final String region;
+ private final String secretId;
+ private final String secretKey;
+ private final String policy;
+ private final Map<String, String> additionInfos;
+
+ public COSSecurityTokenProvider(URI fsUri, Configuration conf) {
+ checkNotNull(fsUri, "fsUri must not be null");
+ this.region = conf.get(REGION);
+ checkNotNull(region, "Region is not set. Please set " + REGION);
+ this.secretId = conf.get(SECRET_ID);
+ checkNotNull(secretId, "Secret ID is not set. Please set " +
SECRET_ID);
+ this.secretKey = conf.get(SECRET_KEY);
+ checkNotNull(secretKey, "Secret Key is not set. Please set " +
SECRET_KEY);
+ this.policy = resolvePolicy(conf, fsUri, region);
+
+ this.additionInfos = new HashMap<>();
+ for (String key : Arrays.asList(REGION, ENDPOINT_KEY)) {
+ if (conf.get(key) != null) {
+ additionInfos.put(key, conf.get(key));
+ }
+ }
+ }
+
+ public ObtainedSecurityToken obtainSecurityToken(String scheme)
+ throws TencentCloudSDKException {
+ LOG.info("Obtaining session credentials token with secret id: {}",
secretId);
+
+ Credential cred = new Credential(secretId, secretKey);
+ StsClient stsClient = new StsClient(cred, region);
+
+ GetFederationTokenRequest request = new GetFederationTokenRequest();
+ request.setName(FEDERATION_TOKEN_NAME);
+ request.setDurationSeconds(DEFAULT_DURATION_SECONDS);
+ request.setPolicy(policy);
+
+ GetFederationTokenResponse response =
stsClient.GetFederationToken(request);
+ com.tencentcloudapi.sts.v20180813.models.Credentials stsCredentials =
+ response.getCredentials();
+
+ // ExpiredTime is a Unix timestamp in seconds, convert to milliseconds
+ long expiredTimeMillis = response.getExpiredTime() * 1000L;
+
+ LOG.info(
+ "Session credentials obtained successfully with tmp secret id:
{}, expiration: {}",
+ stsCredentials.getTmpSecretId(),
+ response.getExpiration());
+
+ return new ObtainedSecurityToken(
+ scheme, toJson(stsCredentials), expiredTimeMillis,
additionInfos);
+ }
+
+ /**
+ * Resolves the STS policy to apply when calling {@code
GetFederationToken}.
+ *
+ * <p>If the user has explicitly configured {@link
+ * org.apache.fluss.fs.cos.COSFileSystemPlugin#SECURITY_TOKEN_POLICY},
that value is used as-is.
+ * Otherwise a default policy is built that scopes the temporary
credential to the bucket and
+ * (optional) prefix derived from the fsUri (i.e. {@code
remote.data.dir}), so the token cannot
+ * be used to access other COS resources.
+ */
+ private static String resolvePolicy(Configuration conf, URI fsUri, String
region) {
+ String userPolicy = conf.get(SECURITY_TOKEN_POLICY);
+ if (userPolicy != null && !userPolicy.trim().isEmpty()) {
+ LOG.info(
+ "Using user-provided STS policy from {} for COS federation
token.",
+ SECURITY_TOKEN_POLICY);
+ return userPolicy;
+ }
+ return buildBucketScopedPolicy(fsUri, region);
+ }
+
+ /**
+ * Builds a default STS policy that grants {@code name/cos:*} only on the
bucket (and optional
+ * key prefix) referenced by the given fsUri.
+ *
+ * <p>COS resource format used here: {@code
qcs::cos:<region>:uid/*:<bucket>/<prefix>*}. The
+ * wildcard owner uid keeps the policy independent of the account uin
while still restricting
+ * access to a specific bucket.
+ */
+ private static String buildBucketScopedPolicy(URI fsUri, String region) {
+ String bucket = fsUri.getAuthority();
+ if (bucket == null || bucket.isEmpty()) {
+ // Fall back to all-resources policy if we cannot derive the
bucket from fsUri.
+ LOG.warn(
+ "Unable to derive bucket from fsUri {}. Falling back to a
wildcard COS resource"
+ + " policy. Consider explicitly configuring {} to
a least-privilege"
+ + " policy.",
+ fsUri,
+ SECURITY_TOKEN_POLICY);
+ return "{"
+ + "\"version\": \"2.0\","
+ + "\"statement\": [{"
+ + "\"action\": [\"name/cos:*\"],"
+ + "\"effect\": \"allow\","
+ + "\"resource\": [\"*\"]"
+ + "}]"
+ + "}";
+ }
+
+ String path = fsUri.getPath();
+ String prefix;
+ if (path == null || path.isEmpty() || "/".equals(path)) {
+ prefix = "";
+ } else {
+ // Strip leading '/' so the resource pattern looks like
"<bucket>/sub/dir/*".
+ prefix = path.startsWith("/") ? path.substring(1) : path;
+ if (!prefix.endsWith("/")) {
+ prefix = prefix + "/";
+ }
+ }
+
+ String resource = "qcs::cos:" + region + ":uid/*:" + bucket + "/" +
prefix + "*";
+ LOG.info("Using bucket-scoped STS policy with resource: {}", resource);
+ return "{"
+ + "\"version\": \"2.0\","
+ + "\"statement\": [{"
+ + "\"action\": [\"name/cos:*\"],"
+ + "\"effect\": \"allow\","
+ + "\"resource\": [\""
+ + resource
+ + "\"]"
+ + "}]"
+ + "}";
+ }
+
+ private byte[] toJson(com.tencentcloudapi.sts.v20180813.models.Credentials
stsCredentials) {
+ Credentials credentials =
+ new Credentials(
+ stsCredentials.getTmpSecretId(),
+ stsCredentials.getTmpSecretKey(),
+ stsCredentials.getToken());
+ return CredentialsJsonSerde.toJson(credentials);
+ }
+}
diff --git
a/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenReceiver.java
b/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenReceiver.java
new file mode 100644
index 000000000..130f375ed
--- /dev/null
+++
b/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenReceiver.java
@@ -0,0 +1,112 @@
+/*
+ * 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.fluss.fs.cos.token;
+
+import org.apache.fluss.fs.cos.COSFileSystemPlugin;
+import org.apache.fluss.fs.token.CredentialsJsonSerde;
+import org.apache.fluss.fs.token.ObtainedSecurityToken;
+import org.apache.fluss.fs.token.SecurityTokenReceiver;
+
+import com.qcloud.cos.auth.BasicSessionCredentials;
+import com.qcloud.cos.auth.COSCredentials;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+import static org.apache.fluss.fs.cos.COSFileSystemPlugin.CREDENTIALS_PROVIDER;
+
+/** Security token receiver for Tencent Cloud COS filesystem. */
+public class COSSecurityTokenReceiver implements SecurityTokenReceiver {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(COSSecurityTokenReceiver.class);
+
+ static volatile COSCredentials credentials;
+ static volatile Map<String, String> additionInfos;
+
+ public static void updateHadoopConfig(org.apache.hadoop.conf.Configuration
hadoopConfig) {
+ updateHadoopConfig(hadoopConfig,
DynamicTemporaryCOSCredentialsProvider.NAME);
+ }
+
+ protected static void updateHadoopConfig(
+ org.apache.hadoop.conf.Configuration hadoopConfig, String
credentialsProviderName) {
+ LOG.info("Updating Hadoop configuration");
+
+ String providers = hadoopConfig.get(CREDENTIALS_PROVIDER, "");
+
+ if (!providers.contains(credentialsProviderName)) {
+ if (providers.isEmpty()) {
+ LOG.debug("Setting provider");
+ providers = credentialsProviderName;
+ } else {
+ providers = credentialsProviderName + "," + providers;
+ LOG.debug("Prepending provider, new providers value: {}",
providers);
+ }
+ hadoopConfig.set(CREDENTIALS_PROVIDER, providers);
+ } else {
+ LOG.debug("Provider already exists");
+ }
+
+ // then, set addition info
+ if (additionInfos == null) {
+ // if addition info is null, it also means we have not received
any token,
+ throw new RuntimeException(
+ "COS credentials have not been received yet. "
+ + "Ensure onNewTokensObtained() has been called
with valid tokens "
+ + "before invoking
COSSecurityTokenReceiver.updateHadoopConfig().");
+ } else {
+ for (Map.Entry<String, String> entry : additionInfos.entrySet()) {
+ hadoopConfig.set(entry.getKey(), entry.getValue());
+ }
+ }
+
+ LOG.info("Updated Hadoop configuration successfully");
+ }
+
+ @Override
+ public String scheme() {
+ return COSFileSystemPlugin.SCHEME;
+ }
+
+ @Override
+ public void onNewTokensObtained(ObtainedSecurityToken token) {
+ LOG.info("Updating session credentials");
+
+ byte[] tokenBytes = token.getToken();
+
+ org.apache.fluss.fs.token.Credentials flussCredentials =
+ CredentialsJsonSerde.fromJson(tokenBytes);
+
+ // STS always returns temporary credentials with a session token
+ credentials =
+ new BasicSessionCredentials(
+ flussCredentials.getAccessKeyId(),
+ flussCredentials.getSecretAccessKey(),
+ flussCredentials.getSecurityToken());
+ additionInfos = token.getAdditionInfos();
+
+ // Consistent with S3DelegationTokenReceiver, logging access key at
INFO level
+ LOG.info(
+ "Session credentials updated successfully with access key:
{}.",
+ credentials.getCOSAccessKeyId());
+ }
+
+ public static COSCredentials getCredentials() {
+ return credentials;
+ }
+}
diff --git
a/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/DynamicTemporaryCOSCredentialsProvider.java
b/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/DynamicTemporaryCOSCredentialsProvider.java
new file mode 100644
index 000000000..71d3e8af8
--- /dev/null
+++
b/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/DynamicTemporaryCOSCredentialsProvider.java
@@ -0,0 +1,69 @@
+/*
+ * 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.fluss.fs.cos.token;
+
+import org.apache.fluss.annotation.Internal;
+
+import com.qcloud.cos.auth.BasicSessionCredentials;
+import com.qcloud.cos.auth.COSCredentials;
+import com.qcloud.cos.auth.COSCredentialsProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Support dynamic credentials for authenticating with Tencent Cloud COS.
It'll get credentials from
+ * {@link COSSecurityTokenReceiver}. It implements COS native {@link
COSCredentialsProvider} to work
+ * with CosNFileSystem.
+ *
+ * <p>This provider supports both static credentials (secretId/secretKey) and
temporary session
+ * credentials (secretId/secretKey/sessionToken).
+ */
+@Internal
+public class DynamicTemporaryCOSCredentialsProvider implements
COSCredentialsProvider {
+
+ private static final Logger LOG =
+
LoggerFactory.getLogger(DynamicTemporaryCOSCredentialsProvider.class);
+
+ public static final String NAME =
DynamicTemporaryCOSCredentialsProvider.class.getName();
+
+ @Override
+ public COSCredentials getCredentials() {
+ COSCredentials credentials = COSSecurityTokenReceiver.getCredentials();
+ if (credentials == null) {
+ throw new RuntimeException(
+ "COS credentials have not been received yet. "
+ + "Ensure COSSecurityTokenReceiver has received
valid tokens.");
+ }
+ if (credentials instanceof BasicSessionCredentials) {
+ BasicSessionCredentials sessionCredentials =
(BasicSessionCredentials) credentials;
+ LOG.debug("Providing session credentials");
+ return new BasicSessionCredentials(
+ sessionCredentials.getCOSAccessKeyId(),
+ sessionCredentials.getCOSSecretKey(),
+ sessionCredentials.getSessionToken());
+ } else {
+ LOG.debug("Providing non-session COS credentials");
+ return credentials;
+ }
+ }
+
+ @Override
+ public void refresh() {
+ // do nothing, credentials are updated by COSSecurityTokenReceiver
+ }
+}
diff --git a/fluss-filesystems/fluss-fs-cos/src/main/resources/META-INF/NOTICE
b/fluss-filesystems/fluss-fs-cos/src/main/resources/META-INF/NOTICE
new file mode 100644
index 000000000..feb58a9a2
--- /dev/null
+++ b/fluss-filesystems/fluss-fs-cos/src/main/resources/META-INF/NOTICE
@@ -0,0 +1,77 @@
+fluss-fs-cos
+Copyright 2025-2026 The Apache Software Foundation
+
+This project includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
+
+This project bundles the following dependencies under the Apache Software
License 2.0 (http://www.apache.org/licenses/LICENSE-2.0.txt)
+
+- com.fasterxml.jackson.core:jackson-annotations:2.15.3
+- com.fasterxml.jackson.core:jackson-core:2.15.3
+- com.fasterxml.jackson.core:jackson-databind:2.15.3
+- com.fasterxml.woodstox:woodstox-core:5.4.0
+- com.google.code.gson:gson:2.8.9
+- com.google.guava:failureaccess:1.0
+- com.google.guava:guava:27.0-jre
+- com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava
+- com.google.j2objc:j2objc-annotations:1.1
+- com.google.re2j:re2j:1.1
+- com.qcloud:cos_api:5.6.139
+- com.squareup.okhttp3:logging-interceptor:4.10.0
+- com.squareup.okhttp3:okhttp:4.10.0
+- com.squareup.okio:okio-jvm:3.2.0
+- com.squareup.okio:okio:3.2.0
+- com.tencentcloudapi:tencentcloud-sdk-java-common:3.1.678
+- com.tencentcloudapi:tencentcloud-sdk-java-kms:3.1.213
+- com.tencentcloudapi:tencentcloud-sdk-java-sts:3.1.678
+- commons-beanutils:commons-beanutils:1.9.4
+- commons-codec:commons-codec:1.13
+- commons-collections:commons-collections:3.2.2
+- commons-io:commons-io:2.14.0
+- commons-logging:commons-logging:1.2
+- dnsjava:dnsjava:3.4.0
+- io.dropwizard.metrics:metrics-core:3.2.4
+- io.netty:netty-buffer:4.1.100.Final
+- io.netty:netty-codec:4.1.100.Final
+- io.netty:netty-common:4.1.100.Final
+- io.netty:netty-handler:4.1.100.Final
+- io.netty:netty-resolver:4.1.100.Final
+- io.netty:netty-transport-classes-epoll:4.1.100.Final
+- io.netty:netty-transport-native-epoll:4.1.100.Final
+- io.netty:netty-transport-native-unix-common:4.1.100.Final
+- io.netty:netty-transport:4.1.100.Final
+- jakarta.activation:jakarta.activation-api:1.2.1
+- joda-time:joda-time:2.9.9
+- org.ini4j:ini4j:0.5.4
+- org.apache.commons:commons-compress:1.24.0
+- org.apache.commons:commons-configuration2:2.8.0
+- org.apache.commons:commons-lang3:3.18.0
+- org.apache.commons:commons-text:1.10.0
+- org.apache.hadoop:hadoop-annotations:3.4.0
+- org.apache.hadoop:hadoop-auth:3.4.0
+- org.apache.hadoop:hadoop-common:3.4.0
+- org.apache.hadoop:hadoop-cos:3.3.5
+- org.apache.hadoop.thirdparty:hadoop-shaded-guava:1.2.0
+- org.apache.hadoop.thirdparty:hadoop-shaded-protobuf_3_21:1.2.0
+- org.apache.httpcomponents:httpclient:4.5.13
+- org.apache.httpcomponents:httpcore:4.4.13
+- org.apache.kerby:kerb-core:2.0.3
+- org.apache.kerby:kerby-asn1:2.0.3
+- org.apache.kerby:kerby-pkix:2.0.3
+- org.apache.kerby:kerby-util:2.0.3
+- org.bouncycastle:bcprov-jdk15on:1.67
+- org.checkerframework:checker-qual:2.5.2
+- org.jetbrains.kotlin:kotlin-stdlib-common:1.6.20
+- org.jetbrains.kotlin:kotlin-stdlib-jdk7:1.6.10
+- org.jetbrains.kotlin:kotlin-stdlib-jdk8:1.6.10
+- org.jetbrains.kotlin:kotlin-stdlib:1.6.20
+- org.jetbrains:annotations:13.0
+- org.codehaus.jettison:jettison:1.5.4
+- org.codehaus.mojo:animal-sniffer-annotations:1.17
+- org.codehaus.woodstox:stax2-api:4.2.1
+- org.xerial.snappy:snappy-java:1.1.10.4
+
+This project bundles the following dependencies under the CDDL 1.1 license.
+See bundled license files for details.
+
+- javax.xml.bind:jaxb-api:2.3.1
diff --git
a/fluss-filesystems/fluss-fs-cos/src/main/resources/META-INF/licenses/LICENSE.jaxb
b/fluss-filesystems/fluss-fs-cos/src/main/resources/META-INF/licenses/LICENSE.jaxb
new file mode 100644
index 000000000..fd16ea954
--- /dev/null
+++
b/fluss-filesystems/fluss-fs-cos/src/main/resources/META-INF/licenses/LICENSE.jaxb
@@ -0,0 +1,135 @@
+COMMON DEVELOPMENT AND DISTRIBUTION LICENSE (CDDL)Version 1.1
+
+1. Definitions.
+
+ 1.1. "Contributor" means each individual or entity that creates or
contributes to the creation of Modifications.
+
+ 1.2. "Contributor Version" means the combination of the Original
Software, prior Modifications used by a Contributor (if any), and the
Modifications made by that particular Contributor.
+
+ 1.3. "Covered Software" means (a) the Original Software, or (b)
Modifications, or (c) the combination of files containing Original Software
with files containing Modifications, in each case including portions thereof.
+
+ 1.4. "Executable" means the Covered Software in any form other than
Source Code.
+
+ 1.5. "Initial Developer" means the individual or entity that first makes
Original Software available under this License.
+
+ 1.6. "Larger Work" means a work which combines Covered Software or
portions thereof with code not governed by the terms of this License.
+
+ 1.7. "License" means this document.
+
+ 1.8. "Licensable" means having the right to grant, to the maximum extent
possible, whether at the time of the initial grant or subsequently acquired,
any and all of the rights conveyed herein.
+
+ 1.9. "Modifications" means the Source Code and Executable form of any of
the following:
+
+ A. Any file that results from an addition to, deletion from or
modification of the contents of a file containing Original Software or previous
Modifications;
+
+ B. Any new file that contains any part of the Original Software or
previous Modification; or
+
+ C. Any new file that is contributed or otherwise made available under the
terms of this License.
+
+ 1.10. "Original Software" means the Source Code and Executable form of
computer software code that is originally released under this License.
+
+ 1.11. "Patent Claims" means any patent claim(s), now owned or hereafter
acquired, including without limitation, method, process, and apparatus claims,
in any patent Licensable by grantor.
+
+ 1.12. "Source Code" means (a) the common form of computer software code
in which modifications are made and (b) associated documentation included in or
with such code.
+
+ 1.13. "You" (or "Your") means an individual or a legal entity exercising
rights under, and complying with all of the terms of, this License. For legal
entities, "You" includes any entity which controls, is controlled by, or is
under common control with You. For purposes of this definition, "control" means
(a) the power, direct or indirect, to cause the direction or management of such
entity, whether by contract or otherwise, or (b) ownership of more than fifty
percent (50%) of the o [...]
+
+2. License Grants.
+
+ 2.1. The Initial Developer Grant.
+
+ Conditioned upon Your compliance with Section 3.1 below and subject to
third party intellectual property claims, the Initial Developer hereby grants
You a world-wide, royalty-free, non-exclusive license:
+
+ (a) under intellectual property rights (other than patent or trademark)
Licensable by Initial Developer, to use, reproduce, modify, display, perform,
sublicense and distribute the Original Software (or portions thereof), with or
without Modifications, and/or as part of a Larger Work; and
+
+ (b) under Patent Claims infringed by the making, using or selling of
Original Software, to make, have made, use, practice, sell, and offer for sale,
and/or otherwise dispose of the Original Software (or portions thereof).
+
+ (c) The licenses granted in Sections 2.1(a) and (b) are effective on the
date Initial Developer first distributes or otherwise makes the Original
Software available to a third party under the terms of this License.
+
+ (d) Notwithstanding Section 2.1(b) above, no patent license is granted:
(1) for code that You delete from the Original Software, or (2) for
infringements caused by: (i) the modification of the Original Software, or (ii)
the combination of the Original Software with other software or devices.
+
+ 2.2. Contributor Grant.
+
+ Conditioned upon Your compliance with Section 3.1 below and subject to
third party intellectual property claims, each Contributor hereby grants You a
world-wide, royalty-free, non-exclusive license:
+
+ (a) under intellectual property rights (other than patent or trademark)
Licensable by Contributor to use, reproduce, modify, display, perform,
sublicense and distribute the Modifications created by such Contributor (or
portions thereof), either on an unmodified basis, with other Modifications, as
Covered Software and/or as part of a Larger Work; and
+
+ (b) under Patent Claims infringed by the making, using, or selling of
Modifications made by that Contributor either alone and/or in combination with
its Contributor Version (or portions of such combination), to make, use, sell,
offer for sale, have made, and/or otherwise dispose of: (1) Modifications made
by that Contributor (or portions thereof); and (2) the combination of
Modifications made by that Contributor with its Contributor Version (or
portions of such combination).
+
+ (c) The licenses granted in Sections 2.2(a) and 2.2(b) are effective on
the date Contributor first distributes or otherwise makes the Modifications
available to a third party.
+
+ (d) Notwithstanding Section 2.2(b) above, no patent license is granted:
(1) for any code that Contributor has deleted from the Contributor Version; (2)
for infringements caused by: (i) third party modifications of Contributor
Version, or (ii) the combination of Modifications made by that Contributor with
other software (except as part of the Contributor Version) or other devices; or
(3) under Patent Claims infringed by Covered Software in the absence of
Modifications made by that Co [...]
+
+3. Distribution Obligations.
+
+ 3.1. Availability of Source Code.
+
+ Any Covered Software that You distribute or otherwise make available in
Executable form must also be made available in Source Code form and that Source
Code form must be distributed only under the terms of this License. You must
include a copy of this License with every copy of the Source Code form of the
Covered Software You distribute or otherwise make available. You must inform
recipients of any such Covered Software in Executable form as to how they can
obtain such Covered Softw [...]
+
+ 3.2. Modifications.
+
+ The Modifications that You create or to which You contribute are governed
by the terms of this License. You represent that You believe Your Modifications
are Your original creation(s) and/or You have sufficient rights to grant the
rights conveyed by this License.
+
+ 3.3. Required Notices.
+
+ You must include a notice in each of Your Modifications that identifies
You as the Contributor of the Modification. You may not remove or alter any
copyright, patent or trademark notices contained within the Covered Software,
or any notices of licensing or any descriptive text giving attribution to any
Contributor or the Initial Developer.
+
+ 3.4. Application of Additional Terms.
+
+ You may not offer or impose any terms on any Covered Software in Source
Code form that alters or restricts the applicable version of this License or
the recipients' rights hereunder. You may choose to offer, and to charge a fee
for, warranty, support, indemnity or liability obligations to one or more
recipients of Covered Software. However, you may do so only on Your own behalf,
and not on behalf of the Initial Developer or any Contributor. You must make it
absolutely clear that any [...]
+
+ 3.5. Distribution of Executable Versions.
+
+ You may distribute the Executable form of the Covered Software under the
terms of this License or under the terms of a license of Your choice, which may
contain terms different from this License, provided that You are in compliance
with the terms of this License and that the license for the Executable form
does not attempt to limit or alter the recipient's rights in the Source Code
form from the rights set forth in this License. If You distribute the Covered
Software in Executable f [...]
+
+ 3.6. Larger Works.
+
+ You may create a Larger Work by combining Covered Software with other
code not governed by the terms of this License and distribute the Larger Work
as a single product. In such a case, You must make sure the requirements of
this License are fulfilled for the Covered Software.
+
+4. Versions of the License.
+
+ 4.1. New Versions.
+
+ Oracle is the initial license steward and may publish revised and/or new
versions of this License from time to time. Each version will be given a
distinguishing version number. Except as provided in Section 4.3, no one other
than the license steward has the right to modify this License.
+
+ 4.2. Effect of New Versions.
+
+ You may always continue to use, distribute or otherwise make the Covered
Software available under the terms of the version of the License under which
You originally received the Covered Software. If the Initial Developer includes
a notice in the Original Software prohibiting it from being distributed or
otherwise made available under any subsequent version of the License, You must
distribute and make the Covered Software available under the terms of the
version of the License under [...]
+
+ 4.3. Modified Versions.
+
+ When You are an Initial Developer and You want to create a new license
for Your Original Software, You may create and use a modified version of this
License if You: (a) rename the license and remove any references to the name of
the license steward (except to note that the license differs from this
License); and (b) otherwise make it clear that the license contains terms which
differ from this License.
+
+5. DISCLAIMER OF WARRANTY.
+
+ COVERED SOFTWARE IS PROVIDED UNDER THIS LICENSE ON AN "AS IS" BASIS,
WITHOUT WARRANTY OF ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING, WITHOUT
LIMITATION, WARRANTIES THAT THE COVERED SOFTWARE IS FREE OF DEFECTS,
MERCHANTABLE, FIT FOR A PARTICULAR PURPOSE OR NON-INFRINGING. THE ENTIRE RISK
AS TO THE QUALITY AND PERFORMANCE OF THE COVERED SOFTWARE IS WITH YOU. SHOULD
ANY COVERED SOFTWARE PROVE DEFECTIVE IN ANY RESPECT, YOU (NOT THE INITIAL
DEVELOPER OR ANY OTHER CONTRIBUTOR) ASSUM [...]
+
+6. TERMINATION.
+
+ 6.1. This License and the rights granted hereunder will terminate
automatically if You fail to comply with terms herein and fail to cure such
breach within 30 days of becoming aware of the breach. Provisions which, by
their nature, must remain in effect beyond the termination of this License
shall survive.
+
+ 6.2. If You assert a patent infringement claim (excluding declaratory
judgment actions) against Initial Developer or a Contributor (the Initial
Developer or Contributor against whom You assert such claim is referred to as
"Participant") alleging that the Participant Software (meaning the Contributor
Version where the Participant is a Contributor or the Original Software where
the Participant is the Initial Developer) directly or indirectly infringes any
patent, then any and all righ [...]
+
+ 6.3. If You assert a patent infringement claim against Participant
alleging that the Participant Software directly or indirectly infringes any
patent where such claim is resolved (such as by license or settlement) prior to
the initiation of patent infringement litigation, then the reasonable value of
the licenses granted by such Participant under Sections 2.1 or 2.2 shall be
taken into account in determining the amount or value of any payment or license.
+
+ 6.4. In the event of termination under Sections 6.1 or 6.2 above, all end
user licenses that have been validly granted by You or any distributor
hereunder prior to termination (excluding licenses granted to You by any
distributor) shall survive termination.
+
+7. LIMITATION OF LIABILITY.
+
+ UNDER NO CIRCUMSTANCES AND UNDER NO LEGAL THEORY, WHETHER TORT (INCLUDING
NEGLIGENCE), CONTRACT, OR OTHERWISE, SHALL YOU, THE INITIAL DEVELOPER, ANY
OTHER CONTRIBUTOR, OR ANY DISTRIBUTOR OF COVERED SOFTWARE, OR ANY SUPPLIER OF
ANY OF SUCH PARTIES, BE LIABLE TO ANY PERSON FOR ANY INDIRECT, SPECIAL,
INCIDENTAL, OR CONSEQUENTIAL DAMAGES OF ANY CHARACTER INCLUDING, WITHOUT
LIMITATION, DAMAGES FOR LOSS OF GOODWILL, WORK STOPPAGE, COMPUTER FAILURE OR
MALFUNCTION, OR ANY AND ALL OTHER COMM [...]
+
+8. U.S. GOVERNMENT END USERS.
+
+ The Covered Software is a "commercial item," as that term is defined in
48 C.F.R. 2.101 (Oct. 1995), consisting of "commercial computer software" (as
that term is defined at 48 C.F.R. ? 252.227-7014(a)(1)) and "commercial
computer software documentation" as such terms are used in 48 C.F.R. 12.212
(Sept. 1995). Consistent with 48 C.F.R. 12.212 and 48 C.F.R. 227.7202-1 through
227.7202-4 (June 1995), all U.S. Government End Users acquire Covered Software
with only those rights set for [...]
+
+9. MISCELLANEOUS.
+
+ This License represents the complete agreement concerning subject matter
hereof. If any provision of this License is held to be unenforceable, such
provision shall be reformed only to the extent necessary to make it
enforceable. This License shall be governed by the law of the jurisdiction
specified in a notice contained within the Original Software (except to the
extent applicable law, if any, provides otherwise), excluding such
jurisdiction's conflict-of-law provisions. Any litiga [...]
+
+10. RESPONSIBILITY FOR CLAIMS.
+
+ As between Initial Developer and the Contributors, each party is
responsible for claims and damages arising, directly or indirectly, out of its
utilization of rights under this License and You agree to work with Initial
Developer and Contributors to distribute such responsibility on an equitable
basis. Nothing herein is intended or shall be deemed to constitute any
admission of liability.
+
+----------
+NOTICE PURSUANT TO SECTION 9 OF THE COMMON DEVELOPMENT AND DISTRIBUTION
LICENSE (CDDL)
+The code released under the CDDL shall be governed by the laws of the State of
California (excluding conflict-of-law provisions). Any litigation relating to
this License shall be subject to the jurisdiction of the Federal Courts of the
Northern District of California and the state courts of the State of
California, with venue lying in Santa Clara County, California.
diff --git
a/fluss-filesystems/fluss-fs-cos/src/main/resources/META-INF/services/org.apache.fluss.fs.FileSystemPlugin
b/fluss-filesystems/fluss-fs-cos/src/main/resources/META-INF/services/org.apache.fluss.fs.FileSystemPlugin
new file mode 100644
index 000000000..3919000bb
--- /dev/null
+++
b/fluss-filesystems/fluss-fs-cos/src/main/resources/META-INF/services/org.apache.fluss.fs.FileSystemPlugin
@@ -0,0 +1,17 @@
+# 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.
+
+org.apache.fluss.fs.cos.COSFileSystemPlugin
diff --git
a/fluss-filesystems/fluss-fs-cos/src/main/resources/META-INF/services/org.apache.fluss.fs.token.SecurityTokenReceiver
b/fluss-filesystems/fluss-fs-cos/src/main/resources/META-INF/services/org.apache.fluss.fs.token.SecurityTokenReceiver
new file mode 100644
index 000000000..262c317c8
--- /dev/null
+++
b/fluss-filesystems/fluss-fs-cos/src/main/resources/META-INF/services/org.apache.fluss.fs.token.SecurityTokenReceiver
@@ -0,0 +1,17 @@
+# 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.
+
+org.apache.fluss.fs.cos.token.COSSecurityTokenReceiver
diff --git
a/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSFileSystemBehaviorITCase.java
b/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSFileSystemBehaviorITCase.java
new file mode 100644
index 000000000..0e00f9f59
--- /dev/null
+++
b/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSFileSystemBehaviorITCase.java
@@ -0,0 +1,67 @@
+/*
+ * 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.fluss.fs.cos;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.fs.FileSystem;
+import org.apache.fluss.fs.FileSystemBehaviorTestSuite;
+import org.apache.fluss.fs.FsPath;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+
+import java.util.UUID;
+
+import static org.apache.fluss.fs.cos.COSFileSystemPlugin.ENDPOINT_KEY;
+import static org.apache.fluss.fs.cos.COSFileSystemPlugin.SECRET_ID;
+import static org.apache.fluss.fs.cos.COSFileSystemPlugin.SECRET_KEY;
+
+/**
+ * An implementation of the {@link FileSystemBehaviorTestSuite} for the COS
file system with Hadoop
+ * cos sdk.
+ */
+class COSFileSystemBehaviorITCase extends FileSystemBehaviorTestSuite {
+
+ private static final String TEST_DATA_DIR = "tests-" + UUID.randomUUID();
+
+ @BeforeAll
+ static void setup() {
+ COSTestCredentials.assumeCredentialsAvailable();
+
+ final Configuration conf = new Configuration();
+ conf.setString(ENDPOINT_KEY,
COSTestCredentials.getCOSEndpointSuffix());
+ conf.setString(SECRET_ID, COSTestCredentials.getCOSSecretId());
+ conf.setString(SECRET_KEY, COSTestCredentials.getCOSSecretKey());
+ FileSystem.initialize(conf, null);
+ }
+
+ @Override
+ protected FileSystem getFileSystem() throws Exception {
+ return getBasePath().getFileSystem();
+ }
+
+ @Override
+ protected FsPath getBasePath() {
+ return new FsPath(COSTestCredentials.getTestBucketUri() +
TEST_DATA_DIR);
+ }
+
+ @AfterAll
+ static void clearFsConfig() {
+ FileSystem.initialize(new Configuration(), null);
+ }
+}
diff --git
a/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSTestCredentials.java
b/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSTestCredentials.java
new file mode 100644
index 000000000..854f4c445
--- /dev/null
+++
b/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSTestCredentials.java
@@ -0,0 +1,119 @@
+/*
+ * 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.fluss.fs.cos;
+
+import javax.annotation.Nullable;
+
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/** Access to credentials to access COS buckets during integration tests. */
+public class COSTestCredentials {
+ @Nullable private static final String ENDPOINT_SUFFIX =
System.getenv("COSN_ENDPOINT_SUFFIX");
+
+ @Nullable private static final String BUCKET =
System.getenv("COSN_BUCKET");
+
+ @Nullable private static final String SECRET_ID =
System.getenv("COSN_SECRET_ID");
+
+ @Nullable private static final String SECRET_KEY =
System.getenv("COSN_SECRET_KEY");
+
+ @Nullable private static final String REGION =
System.getenv("COSN_REGION");
+
+ // ------------------------------------------------------------------------
+
+ public static boolean credentialsAvailable() {
+ return isNotEmpty(ENDPOINT_SUFFIX)
+ && isNotEmpty(BUCKET)
+ && isNotEmpty(SECRET_ID)
+ && isNotEmpty(SECRET_KEY)
+ && isNotEmpty(REGION);
+ }
+
+ /** Checks if a String is not null and not empty. */
+ private static boolean isNotEmpty(@Nullable String str) {
+ return str != null && !str.isEmpty();
+ }
+
+ public static void assumeCredentialsAvailable() {
+ assumeTrue(
+ credentialsAvailable(), "No COS credentials available in this
test's environment");
+ }
+
+ /**
+ * Get COS endpoint suffix used to connect.
+ *
+ * @return COS endpoint suffix
+ */
+ public static String getCOSEndpointSuffix() {
+ if (ENDPOINT_SUFFIX != null) {
+ return ENDPOINT_SUFFIX;
+ } else {
+ throw new IllegalStateException("COS endpoint suffix is not
available");
+ }
+ }
+
+ /**
+ * Get COS secret id.
+ *
+ * @return COS secret id
+ */
+ public static String getCOSSecretId() {
+ if (SECRET_ID != null) {
+ return SECRET_ID;
+ } else {
+ throw new IllegalStateException("COS secret id is not available");
+ }
+ }
+
+ /**
+ * Get COS secret key.
+ *
+ * @return COS secret key
+ */
+ public static String getCOSSecretKey() {
+ if (SECRET_KEY != null) {
+ return SECRET_KEY;
+ } else {
+ throw new IllegalStateException("COS secret key is not available");
+ }
+ }
+
+ /**
+ * Get COS region.
+ *
+ * @return COS region
+ */
+ public static String getCOSRegion() {
+ if (REGION != null) {
+ return REGION;
+ } else {
+ throw new IllegalStateException("COS region is not available");
+ }
+ }
+
+ public static String getTestBucketUri() {
+ return getTestBucketUriWithScheme("cosn");
+ }
+
+ public static String getTestBucketUriWithScheme(String scheme) {
+ if (BUCKET != null) {
+ return scheme + "://" + BUCKET + "/";
+ } else {
+ throw new IllegalStateException("COS test bucket is not
available");
+ }
+ }
+}
diff --git
a/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSWithCredentialsProviderFileSystemBehaviorITCase.java
b/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSWithCredentialsProviderFileSystemBehaviorITCase.java
new file mode 100644
index 000000000..9582ae560
--- /dev/null
+++
b/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSWithCredentialsProviderFileSystemBehaviorITCase.java
@@ -0,0 +1,68 @@
+/*
+ * 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.fluss.fs.cos;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.fs.FileSystem;
+import org.apache.fluss.fs.FileSystemBehaviorTestSuite;
+import org.apache.fluss.fs.FsPath;
+
+import com.qcloud.cos.auth.COSCredentialsProvider;
+import org.apache.hadoop.fs.cosn.auth.EnvironmentVariableCredentialsProvider;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+
+import java.util.UUID;
+
+import static org.apache.fluss.fs.cos.COSFileSystemPlugin.CREDENTIALS_PROVIDER;
+import static org.apache.fluss.fs.cos.COSFileSystemPlugin.ENDPOINT_KEY;
+
+/** IT case for access cos via set {@link COSCredentialsProvider}. */
+class COSWithCredentialsProviderFileSystemBehaviorITCase extends
FileSystemBehaviorTestSuite {
+
+ private static final String TEST_DATA_DIR = "tests-" + UUID.randomUUID();
+
+ @BeforeAll
+ static void setup() {
+ COSTestCredentials.assumeCredentialsAvailable();
+
+ // Use EnvironmentVariableCredentialsProvider shipped with hadoop-cos,
which reads
+ // credentials from the COSN_SECRET_ID / COSN_SECRET_KEY environment
variables
+ // (the same ones already required by COSTestCredentials).
+ final Configuration conf = new Configuration();
+ conf.setString(ENDPOINT_KEY,
COSTestCredentials.getCOSEndpointSuffix());
+ conf.setString(
+ CREDENTIALS_PROVIDER,
EnvironmentVariableCredentialsProvider.class.getName());
+ FileSystem.initialize(conf, null);
+ }
+
+ @Override
+ protected FileSystem getFileSystem() throws Exception {
+ return getBasePath().getFileSystem();
+ }
+
+ @Override
+ protected FsPath getBasePath() {
+ return new FsPath(COSTestCredentials.getTestBucketUri() +
TEST_DATA_DIR);
+ }
+
+ @AfterAll
+ static void clearFsConfig() {
+ FileSystem.initialize(new Configuration(), null);
+ }
+}
diff --git
a/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSWithTokenFileSystemBehaviorBaseITCase.java
b/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSWithTokenFileSystemBehaviorBaseITCase.java
new file mode 100644
index 000000000..46586c0f6
--- /dev/null
+++
b/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSWithTokenFileSystemBehaviorBaseITCase.java
@@ -0,0 +1,43 @@
+/*
+ * 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.fluss.fs.cos;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.fs.FileSystem;
+import org.apache.fluss.fs.FileSystemBehaviorTestSuite;
+
+import static org.apache.fluss.fs.cos.COSFileSystemPlugin.ENDPOINT_KEY;
+import static org.apache.fluss.fs.cos.COSFileSystemPlugin.REGION;
+import static org.apache.fluss.fs.cos.COSFileSystemPlugin.SECRET_ID;
+import static org.apache.fluss.fs.cos.COSFileSystemPlugin.SECRET_KEY;
+
+/** Base IT case for access COS with temporary credentials in hadoop sdk as
COS FileSystem. */
+abstract class COSWithTokenFileSystemBehaviorBaseITCase extends
FileSystemBehaviorTestSuite {
+
+ static void initFileSystemWithSecretKey() {
+ COSTestCredentials.assumeCredentialsAvailable();
+
+ // first init filesystem with secretId/secretKey
+ final Configuration conf = new Configuration();
+ conf.setString(ENDPOINT_KEY,
COSTestCredentials.getCOSEndpointSuffix());
+ conf.setString(REGION, COSTestCredentials.getCOSRegion());
+ conf.setString(SECRET_ID, COSTestCredentials.getCOSSecretId());
+ conf.setString(SECRET_KEY, COSTestCredentials.getCOSSecretKey());
+ FileSystem.initialize(conf, null);
+ }
+}
diff --git
a/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSWithTokenFileSystemBehaviorITCase.java
b/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSWithTokenFileSystemBehaviorITCase.java
new file mode 100644
index 000000000..c5dd30c86
--- /dev/null
+++
b/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSWithTokenFileSystemBehaviorITCase.java
@@ -0,0 +1,72 @@
+/*
+ * 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.fluss.fs.cos;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.fs.FileSystem;
+import org.apache.fluss.fs.FsPath;
+import org.apache.fluss.fs.cos.token.COSSecurityTokenReceiver;
+import org.apache.fluss.fs.token.ObtainedSecurityToken;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+
+import java.util.UUID;
+
+/** IT case for access COS with security token in hadoop sdk as FileSystem. */
+class COSWithTokenFileSystemBehaviorITCase extends
COSWithTokenFileSystemBehaviorBaseITCase {
+
+ private static final String TEST_DATA_DIR = "tests-" + UUID.randomUUID();
+
+ @BeforeAll
+ static void setup() throws Exception {
+ // init a filesystem with secretId/secretKey so that it can generate
token
+ initFileSystemWithSecretKey();
+ // now, we can init with token
+ initFileSystemWithToken(getFsPath());
+ }
+
+ @Override
+ protected FileSystem getFileSystem() throws Exception {
+ return getFsPath().getFileSystem();
+ }
+
+ @Override
+ protected FsPath getBasePath() {
+ return getFsPath();
+ }
+
+ protected static FsPath getFsPath() {
+ return new FsPath(COSTestCredentials.getTestBucketUri() +
TEST_DATA_DIR);
+ }
+
+ @AfterAll
+ static void clearFsConfig() {
+ FileSystem.initialize(new Configuration(), null);
+ }
+
+ private static void initFileSystemWithToken(FsPath fsPath) throws
Exception {
+ Configuration configuration = new Configuration();
+ // obtain a security token and call onNewTokensObtained
+ ObtainedSecurityToken obtainedSecurityToken =
fsPath.getFileSystem().obtainSecurityToken();
+ COSSecurityTokenReceiver cosSecurityTokenReceiver = new
COSSecurityTokenReceiver();
+ cosSecurityTokenReceiver.onNewTokensObtained(obtainedSecurityToken);
+
+ FileSystem.initialize(configuration, null);
+ }
+}
diff --git a/fluss-filesystems/pom.xml b/fluss-filesystems/pom.xml
index a3bafb480..34a81de98 100644
--- a/fluss-filesystems/pom.xml
+++ b/fluss-filesystems/pom.xml
@@ -36,6 +36,7 @@
<module>fluss-fs-gs</module>
<module>fluss-fs-azure</module>
<module>fluss-fs-obs</module>
+ <module>fluss-fs-cos</module>
<module>fluss-fs-hdfs</module>
</modules>
<packaging>pom</packaging>
diff --git a/fluss-test-coverage/pom.xml b/fluss-test-coverage/pom.xml
index 2f9fdba58..944378b00 100644
--- a/fluss-test-coverage/pom.xml
+++ b/fluss-test-coverage/pom.xml
@@ -429,6 +429,7 @@
</exclude>
<exclude>org.apache.fluss.fs.hdfs.HdfsSecurityTokenReceiver</exclude>
<exclude>org.apache.fluss.fs.oss.*</exclude>
+
<exclude>org.apache.fluss.fs.cos.*</exclude>
<exclude>org.apache.fluss.fs.s3.*</exclude>
<exclude>org.apache.fluss.fs.obs.*</exclude>
<exclude>com.amazonaws.services.s3.model.transform.XmlResponsesSaxParser*
diff --git a/website/docs/maintenance/filesystems/cos.md
b/website/docs/maintenance/filesystems/cos.md
new file mode 100644
index 000000000..192c4cbd3
--- /dev/null
+++ b/website/docs/maintenance/filesystems/cos.md
@@ -0,0 +1,89 @@
+---
+title: Tencent Cloud COS
+sidebar_position: 7
+---
+
+# Tencent Cloud COS
+
+[Tencent Cloud Object Storage](https://cloud.tencent.com/product/cos) (Tencent
Cloud COS) is a distributed storage service offered by Tencent Cloud for
storing massive amounts of data. It provides high scalability, low cost,
reliability, and security.
+
+## Install COS Plugin Manually
+
+Tencent Cloud COS support is not included in the default Fluss distribution.
To enable COS support, you need to manually install the filesystem plugin into
Fluss.
+
+1. **Prepare the plugin JAR**:
+
+ - Download the `fluss-fs-cos-$FLUSS_VERSION$.jar` from the [Maven
Repository](https://repo1.maven.org/maven2/org/apache/fluss/fluss-fs-cos/$FLUSS_VERSION$/fluss-fs-cos-$FLUSS_VERSION$.jar).
+
+2. **Place the plugin**: Place the plugin JAR file in the
`${FLUSS_HOME}/plugins/cos/` directory:
+ ```bash
+ mkdir -p ${FLUSS_HOME}/plugins/cos/
+ cp fluss-fs-cos-$FLUSS_VERSION$.jar ${FLUSS_HOME}/plugins/cos/
+ ```
+
+3. Restart Fluss if the cluster is already running to ensure the new plugin is
loaded.
+
+## Configurations setup
+
+To enable Tencent Cloud COS as remote storage, there are some required
configurations that must be added to Fluss' `server.yaml`:
+
+```yaml
+# The dir that used to be as the remote storage of Fluss
+remote.data.dir: cosn://<your-bucket>/path/to/remote/storage
+# COS endpoint suffix, such as: cos.ap-guangzhou.myqcloud.com
+fs.cosn.bucket.endpoint_suffix: <your-endpoint-suffix>
+# COS region, such as: ap-guangzhou
+fs.cosn.userinfo.region: <your-cos-region>
+
+# Authentication (choose one option below)
+
+# Option 1: Direct credentials
+# Tencent Cloud secret id
+fs.cosn.userinfo.secretId: <your-secret-id>
+# Tencent Cloud secret key
+fs.cosn.userinfo.secretKey: <your-secret-key>
+
+# Option 2: Secure credential provider
+fs.cosn.credentials.provider: <your-credentials-provider>
+```
+To avoid exposing sensitive access key information directly in the
`server.yaml`, you can choose option2 to use a credential provider by setting
the `fs.cosn.credentials.provider` property.
+
+For example, to use environment variables for credential management:
+```yaml
+fs.cosn.credentials.provider:
org.apache.hadoop.fs.cosn.auth.EnvironmentVariableCredentialsProvider
+```
+Then, set the following environment variables before starting the Fluss
service:
+```bash
+export COSN_SECRET_ID=<your-secret-id>
+export COSN_SECRET_KEY=<your-secret-key>
+```
+This approach enhances security by keeping sensitive credentials out of
configuration files.
+
+## Token-based Authentication
+
+For client to access the remote storage such as reading snapshot or tiered
log, client must obtain a temporary credential (STS token) from Fluss cluster.
+
+Fluss uses Tencent Cloud STS
[GetFederationToken](https://cloud.tencent.com/document/product/1312/48195) API
to obtain temporary credentials.
+Unlike `AssumeRole`, `GetFederationToken` does not require a role ARN — it
generates temporary credentials directly from the caller's permanent
credentials (`secretId` / `secretKey`).
+
+To enable this, you must configure `fs.cosn.userinfo.region` which is required
for calling the STS API.
+The `fs.cosn.userinfo.region` specifies the COS region where your bucket is
located, such as `ap-guangzhou`, `ap-beijing`, etc. You can
+find different regions in [Tencent Cloud COS
Regions](https://cloud.tencent.com/document/product/436/6224).
+
+### How it works
+
+1. The Fluss server calls `GetFederationToken` with the configured permanent
credentials to obtain a temporary credential consisting of a temporary secret
id, a temporary secret key, and a session token.
+2. By default, the temporary credential is granted **full COS read-write
permission** (`name/cos:*`) scoped to the bucket configured via
`remote.data.dir`, and has a **validity period of 1 hour**. You can override
this default policy by setting `fs.cosn.security.token.policy` to a custom STS
policy JSON.
+3. The temporary credential is distributed to clients, which use it to access
COS directly.
+4. Fluss automatically refreshes the credential before it expires.
+
+:::note
+The effective permission of the temporary credential is the **intersection**
of the permanent credential's permission and the policy specified in the
`GetFederationToken` request.
+So the permanent credential (the `secretId` / `secretKey` configured in
`server.yaml`) must have sufficient COS permissions.
+See more details in [Tencent Cloud COS Access
Policy](https://cloud.tencent.com/document/product/436/6884).
+:::
+
+## Advanced Configurations
+
+Apart from the above configurations, you can also define the configuration
keys mentioned in the [Hadoop-COS
documentation](https://hadoop.apache.org/docs/current/hadoop-cos/cloud-storage/index.html)
+in the Fluss' `server.yaml`. These configurations defined in Hadoop-COS
documentation are advanced configurations which are usually used by performance
tuning.
diff --git a/website/docs/maintenance/filesystems/overview.md
b/website/docs/maintenance/filesystems/overview.md
index 17e4adeb4..c02abbda6 100644
--- a/website/docs/maintenance/filesystems/overview.md
+++ b/website/docs/maintenance/filesystems/overview.md
@@ -7,7 +7,7 @@ sidebar_position: 1
# File Systems
Fluss uses file systems as remote storage to store snapshots for Primary-Key
Table and store tiered log segments for Log Table. These
-are some of the file systems that Fluss supports currently, including *local*,
*hadoop*, *Aliyun OSS*.
+are some of the file systems that Fluss supports currently, including *local*,
*hadoop*, *Aliyun OSS*, *Tencent Cloud COS*.
The file system used for a particular file is determined by its URI scheme.
For example, `file:///home/user/text.txt` refers to a file in the local file
system,
while `hdfs://namenode:50010/data/user/text.txt` is a file in a specific HDFS
cluster.
@@ -40,4 +40,6 @@ The Fluss project supports the following file systems:
- **[HuaweiCloud OBS](obs.md)** is supported by `fluss-fs-obs` and registered
under the `obs://` URI scheme. Please make sure to [manually install the OBS
plugin](obs.md#install-obs-plugin-manually).
+- **[Tencent Cloud COS](cos.md)** is supported by `fluss-fs-cos` and
registered under the `cosn://` URI scheme. Please make sure to [manually
install the COS plugin](cos.md#install-cos-plugin-manually).
+
The implementation is based on [Hadoop Project](https://hadoop.apache.org/)
but is self-contained with no dependency footprint.
\ No newline at end of file
