[FLUSS-2686] Add COS filesystem support

Author: XuQianJin-Stars

fluss-filesystems/fluss-fs-cos/pom.xml

@@ -121,10 +121,17 @@
                 <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>
@@ -237,17 +244,6 @@
                 </executions>
             </plugin>
 
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-jar-plugin</artifactId>
-                <executions>
-                    <execution>
-                        <goals>
-                            <goal>test-jar</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
         </plugins>
     </build>
 

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

@@ -22,6 +22,7 @@
 import org.apache.fluss.fs.token.ObtainedSecurityToken;
 import org.apache.fluss.fs.token.SecurityTokenReceiver;
 
+import com.qcloud.cos.auth.BasicCOSCredentials;
 import com.qcloud.cos.auth.BasicSessionCredentials;
 import com.qcloud.cos.auth.COSCredentials;
 import org.slf4j.Logger;
@@ -65,7 +66,10 @@ protected static void updateHadoopConfig(
         // then, set addition info
         if (additionInfos == null) {
             // if addition info is null, it also means we have not received any token,
-            throw new RuntimeException("Credentials is not ready.");
+            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());
@@ -89,14 +93,23 @@ public void onNewTokensObtained(ObtainedSecurityToken token) {
         org.apache.fluss.fs.token.Credentials flussCredentials =
                 CredentialsJsonSerde.fromJson(tokenBytes);
 
-        // Create Credential from fluss credentials
-        credentials =
-                new BasicSessionCredentials(
-                        flussCredentials.getAccessKeyId(),
-                        flussCredentials.getSecretAccessKey(),
-                        flussCredentials.getSecurityToken());
+        // Create COSCredentials from fluss credentials, distinguishing between
+        // static credentials (no session token) and temporary session credentials
+        if (flussCredentials.getSecurityToken() != null) {
+            credentials =
+                    new BasicSessionCredentials(
+                            flussCredentials.getAccessKeyId(),
+                            flussCredentials.getSecretAccessKey(),
+                            flussCredentials.getSecurityToken());
+        } else {
+            credentials =
+                    new BasicCOSCredentials(
+                            flussCredentials.getAccessKeyId(),
+                            flussCredentials.getSecretAccessKey());
+        }
         additionInfos = token.getAdditionInfos();
 
+        // Consistent with S3DelegationTokenReceiver, logging access key at INFO level
         LOG.info(
                 "Session credentials updated successfully with access key: {}.",
                 credentials.getCOSAccessKeyId());

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

@@ -26,9 +26,12 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * Support dynamic session credentials for authenticating with Tencent Cloud COS. It'll get
- * credentials from {@link COSSecurityTokenReceiver}. It implements COS native {@link
- * COSCredentialsProvider} to work with CosNFileSystem.
+ * 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 {
@@ -42,13 +45,21 @@ public class DynamicTemporaryCOSCredentialsProvider implements COSCredentialsPro
     public COSCredentials getCredentials() {
         COSCredentials credentials = COSSecurityTokenReceiver.getCredentials();
         if (credentials == null) {
-            throw new RuntimeException("Credentials is not ready.");
+            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;
         }
-        LOG.debug("Providing session credentials");
-        return new BasicSessionCredentials(
-                credentials.getCOSAccessKeyId(),
-                credentials.getCOSSecretKey(),
-                ((BasicSessionCredentials) credentials).getSessionToken());
     }
 
     @Override