[FLUSS-2686] Add COS filesystem support by XuQianJin-Stars · Pull Request #2836 · apache/fluss

XuQianJin-Stars · 2026-03-10T11:02:15Z

Purpose

Linked issue: close #2686

This PR adds support for Tencent Cloud Object Storage (COS) as a remote filesystem for Fluss. COS is one of the most widely used cloud storage services in China and this integration enables users to store Fluss remote data on COS, similar to the existing OSS, OBS, S3, and GCS filesystem integrations.

Brief change log

fluss-fs-cos/pom.xml: Add module configuration with dependencies on hadoop-cos (3.3.5), cos_api (5.6.139), and fluss-fs-hadoop for Hadoop filesystem abstraction.
COSFileSystem: Extends HadoopFileSystem to wrap Hadoop's CosNFileSystem, with lazy-initialized security token provider support for obtaining temporary credentials.
COSFileSystemPlugin: Implements FileSystemPlugin SPI for the cosn scheme. Handles Hadoop configuration translation from Fluss config (prefix fs.cosn.*), supports three credential modes: static secret key, custom credentials provider, and dynamic security token via COSSecurityTokenReceiver.
COSSecurityTokenProvider: Generates temporary security tokens using Tencent Cloud STS (Security Token Service) based on existing secretId/secretKey configuration.
COSSecurityTokenReceiver: Implements SecurityTokenReceiver SPI to receive and apply security tokens, configuring DynamicTemporaryCOSCredentialsProvider as the Hadoop credentials provider.
DynamicTemporaryCOSCredentialsProvider: Implements COS SDK's COSCredentialsProvider interface, providing BasicSessionCredentials from tokens received via COSSecurityTokenReceiver.
fluss-filesystems/pom.xml: Register fluss-fs-cos as a sub-module.

Tests

COSFileSystemBehaviorITCase: Integration test for basic COS filesystem behavior (create, read, write, delete) using static credentials (secretId/secretKey).
COSWithTokenFileSystemBehaviorITCase: Integration test for COS filesystem behavior using dynamic security tokens obtained via STS.
COSWithTokenFileSystemBehaviorBaseITCase: Base class for token-based filesystem tests, handling filesystem initialization with both static credentials and security tokens.
COSTestCredentials: Test utility that reads COS credentials and endpoint from environment variables (COSN_SECRET_ID, COSN_SECRET_KEY, COSN_ENDPOINT, COSN_BUCKET).

Note: These are IT (integration tests) that require actual COS credentials and bucket access. They are skipped automatically when credentials are not available via Assumptions.assumeTrue.

API and Format

No existing API or storage format changes. This PR only adds a new filesystem plugin that registers via SPI (META-INF/services/org.apache.fluss.fs.FileSystemPlugin).

New configuration keys introduced (all following standard Hadoop COS conventions):

fs.cosn.endpoint — COS region endpoint (e.g., ap-guangzhou)
fs.cosn.userinfo.secretId — Tencent Cloud secret ID
fs.cosn.userinfo.secretKey — Tencent Cloud secret key
fs.cosn.credentials.provider — Custom credentials provider class

Documentation

This PR introduces a new feature. Documentation for COS filesystem configuration should be added to the Fluss documentation site in a follow-up. Usage example:

remote.data.dir: cosn://bucket-name/fluss-data
fs.cosn.endpoint: ap-guangzhou
fs.cosn.userinfo.secretId: <your-secret-id>
fs.cosn.userinfo.secretKey: <your-secret-key>

Copilot

Pull request overview

Adds a new Fluss filesystem plugin module to support Tencent Cloud Object Storage (COS) via Hadoop’s CosNFileSystem, including SPI registration and integration tests for COS access.

Changes:

Introduces new fluss-fs-cos module with COS filesystem implementation, token receiver/provider, and SPI service registrations.
Adds COS integration tests (static credentials + token-based flow) gated by environment-provided credentials.
Updates parent filesystem build to include the COS module and adjusts test-coverage exclusions.

Reviewed changes

Copilot reviewed 16 out of 16 changed files in this pull request and generated 10 comments.

Show a summary per file

File	Description
fluss-test-coverage/pom.xml	Excludes COS filesystem packages from coverage aggregation (consistent with other FS plugins).
fluss-filesystems/pom.xml	Registers `fluss-fs-cos` as a submodule.
fluss-filesystems/fluss-fs-cos/pom.xml	Adds COS module dependencies, shading, and multi-release JAXB handling.
fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/COSFileSystem.java	Wraps Hadoop FS and adds Fluss security-token acquisition hook.
fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/COSFileSystemPlugin.java	Implements `FileSystemPlugin` for `cosn` and translates Fluss config to Hadoop config.
fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProvider.java	Provides a Fluss `ObtainedSecurityToken` intended for COS credential propagation.
fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenReceiver.java	Receives Fluss tokens and configures Hadoop COS credential provider + extra config.
fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/DynamicTemporaryCOSCredentialsProvider.java	Supplies COS SDK credentials dynamically from received tokens.
fluss-filesystems/fluss-fs-cos/src/main/resources/META-INF/services/org.apache.fluss.fs.FileSystemPlugin	SPI registration of `COSFileSystemPlugin`.
fluss-filesystems/fluss-fs-cos/src/main/resources/META-INF/services/org.apache.fluss.fs.token.SecurityTokenReceiver	SPI registration of `COSSecurityTokenReceiver`.
fluss-filesystems/fluss-fs-cos/src/main/resources/META-INF/licenses/LICENSE.jaxb	Bundled CDDL license text for JAXB dependency.
fluss-filesystems/fluss-fs-cos/src/main/resources/META-INF/NOTICE	Declares bundled dependencies and licensing notice for the module.
fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSFileSystemBehaviorITCase.java	COS behavior IT using static `secretId/secretKey`.
fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSWithTokenFileSystemBehaviorBaseITCase.java	Base for token-based COS behavior tests; initializes FS with static creds first.
fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSWithTokenFileSystemBehaviorITCase.java	Token-based COS behavior IT wiring receiver + re-init.
fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/COSTestCredentials.java	Reads COS IT credentials from environment variables and gates tests.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-03-11T14:39:42Z

...stems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProvider.java

+    public ObtainedSecurityToken obtainSecurityToken(String scheme) {
+        // For COS, we directly use the configured secret id and secret key as the token.
+        // If STS temporary credentials are needed in the future, this can be extended
+        // to call Tencent Cloud STS API to get temporary credentials.
+        Map<String, String> additionInfo = new HashMap<>();
+        // we need to put endpoint as addition info
+        if (endpoint != null) {
+            additionInfo.put(ENDPOINT_KEY, endpoint);
+        }
+
+        Credentials credentials = new Credentials(secretId, secretKey, null);
+        byte[] tokenBytes = CredentialsJsonSerde.toJson(credentials);
+
+        // token does not expire when using static credentials
+        return new ObtainedSecurityToken(scheme, tokenBytes, Long.MAX_VALUE, additionInfo);
+    }


This provider serializes secretId/secretKey with a null security token and sets Long.MAX_VALUE expiry, but the receiver converts tokens into BasicSessionCredentials (session-based) which generally expects a non-null session token. Either (a) actually obtain STS temporary credentials (accessKey/secretKey/sessionToken + real expiry) and serialize them, or (b) treat the propagated token as static credentials end-to-end (adjust receiver/provider to use non-session COS credentials) and rename/comment accordingly to avoid implying STS behavior.

Copilot · 2026-03-11T14:39:42Z

...-cos/src/main/java/org/apache/fluss/fs/cos/token/DynamicTemporaryCOSCredentialsProvider.java

+        LOG.debug("Providing session credentials");
+        return new BasicSessionCredentials(
+                credentials.getCOSAccessKeyId(),
+                credentials.getCOSSecretKey(),
+                ((BasicSessionCredentials) credentials).getSessionToken());


This unconditionally casts credentials to BasicSessionCredentials, which will throw ClassCastException if the receiver ever stores a different COSCredentials implementation (e.g., static/basic credentials). Consider removing the cast by storing the session token separately, or branching with instanceof and returning the appropriate credential type based on what was received.

Suggested change

LOG.debug("Providing session credentials");

return new BasicSessionCredentials(

credentials.getCOSAccessKeyId(),

credentials.getCOSSecretKey(),

((BasicSessionCredentials) credentials).getSessionToken());

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;

}

Copilot · 2026-03-11T14:39:43Z