Copilot commented on code in PR #2836: URL: https://github.com/apache/fluss/pull/2836#discussion_r2918821915
########## fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProvider.java: ########## @@ -0,0 +1,63 @@ +/* + * 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 org.apache.hadoop.conf.Configuration; + +import java.util.HashMap; +import java.util.Map; + +import static org.apache.fluss.fs.cos.COSFileSystemPlugin.ENDPOINT_KEY; + +/** A provider to provide Tencent Cloud COS security token. */ +public class COSSecurityTokenProvider { + + private static final String SECRET_ID = "fs.cosn.userinfo.secretId"; + private static final String SECRET_KEY = "fs.cosn.userinfo.secretKey"; + + private final String endpoint; + private final String secretId; + private final String secretKey; + + public COSSecurityTokenProvider(Configuration conf) { + endpoint = conf.get(ENDPOINT_KEY); + secretId = conf.get(SECRET_ID); + secretKey = conf.get(SECRET_KEY); + } + + 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); + } Review Comment: 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. ########## fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenReceiver.java: ########## @@ -0,0 +1,108 @@ +/* + * 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("Credentials is not ready."); + } 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); + + // Create Credential from fluss credentials + credentials = + new BasicSessionCredentials( + flussCredentials.getAccessKeyId(), + flussCredentials.getSecretAccessKey(), + flussCredentials.getSecurityToken()); + additionInfos = token.getAdditionInfos(); + + LOG.info( + "Session credentials updated successfully with access key: {}.", + credentials.getCOSAccessKeyId()); Review Comment: Avoid logging credential identifiers (even access key IDs) at INFO level, as they are typically treated as sensitive and can leak via centralized logs. Consider removing the access key from the message, masking it (e.g., last 4 chars), and/or lowering the log level to DEBUG. ```suggestion LOG.info("Session credentials updated successfully."); ``` ########## fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenReceiver.java: ########## @@ -0,0 +1,108 @@ +/* + * 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("Credentials is not ready."); Review Comment: The exception message is grammatically incorrect and not very actionable. Consider using a more specific message that indicates the real cause (e.g., token/credentials not yet received via `onNewTokensObtained`, so the receiver cannot update Hadoop config) and—if applicable—include what to configure/call first. ```suggestion throw new RuntimeException( "Cannot update Hadoop configuration: COS credentials and additional " + "information have not been received yet. Ensure " + "onNewTokensObtained(...) has been called with valid tokens " + "before invoking COSSecurityTokenReceiver.updateHadoopConfig()."); ``` ########## fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenReceiver.java: ########## @@ -0,0 +1,108 @@ +/* + * 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"); + } Review Comment: Using `String#contains` on a comma-separated provider list can yield false positives (e.g., partial class-name matches). Consider parsing the list as comma-separated tokens (trimmed) and checking for exact equality before prepending. ########## fluss-filesystems/fluss-fs-cos/pom.xml: ########## @@ -0,0 +1,254 @@ +<?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>0.10-SNAPSHOT</version> + </parent> + + <artifactId>fluss-fs-cos</artifactId> + <name>Fluss : FileSystems : COS FS</name> + + <properties> + <fs.cosn.sdk.version>3.3.5</fs.cosn.sdk.version> + <fs.cos.api.version>5.6.139</fs.cos.api.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.cosn.sdk.version}</version> + <exclusions> + <exclusion> + <groupId>com.qcloud</groupId> + <artifactId>cos_api</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> + <!-- 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> + <goals> + <goal>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> Review Comment: The same `maven-jar-plugin` is declared twice. This is easy to misread and can lead to configuration being overridden unexpectedly depending on Maven’s model merging. Consider merging the `jar` and `test-jar` executions into a single `maven-jar-plugin` declaration so the `Multi-Release` manifest configuration applies consistently. ########## fluss-filesystems/fluss-fs-cos/pom.xml: ########## @@ -0,0 +1,254 @@ +<?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>0.10-SNAPSHOT</version> + </parent> + + <artifactId>fluss-fs-cos</artifactId> + <name>Fluss : FileSystems : COS FS</name> + + <properties> + <fs.cosn.sdk.version>3.3.5</fs.cosn.sdk.version> + <fs.cos.api.version>5.6.139</fs.cos.api.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.cosn.sdk.version}</version> + <exclusions> + <exclusion> + <groupId>com.qcloud</groupId> + <artifactId>cos_api</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> + <!-- 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> + <goals> + <goal>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> + + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-jar-plugin</artifactId> + <executions> + <execution> + <goals> + <goal>test-jar</goal> + </goals> + </execution> + </executions> + </plugin> Review Comment: The same `maven-jar-plugin` is declared twice. This is easy to misread and can lead to configuration being overridden unexpectedly depending on Maven’s model merging. Consider merging the `jar` and `test-jar` executions into a single `maven-jar-plugin` declaration so the `Multi-Release` manifest configuration applies consistently. ########## fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/DynamicTemporaryCOSCredentialsProvider.java: ########## @@ -0,0 +1,58 @@ +/* + * 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 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. + */ +@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("Credentials is not ready."); + } + LOG.debug("Providing session credentials"); + return new BasicSessionCredentials( + credentials.getCOSAccessKeyId(), + credentials.getCOSSecretKey(), + ((BasicSessionCredentials) credentials).getSessionToken()); Review Comment: 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. ```suggestion 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; } ``` ########## fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenReceiver.java: ########## @@ -0,0 +1,108 @@ +/* + * 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) { Review Comment: The field name `additionInfos` is awkward/unclear; `additionalInfos` (or `additionalInfo` if singular) is more standard and improves readability throughout this class. ########## fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenReceiver.java: ########## @@ -0,0 +1,108 @@ +/* + * 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("Credentials is not ready."); + } else { + for (Map.Entry<String, String> entry : additionInfos.entrySet()) { Review Comment: The field name `additionInfos` is awkward/unclear; `additionalInfos` (or `additionalInfo` if singular) is more standard and improves readability throughout this class. ########## fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenReceiver.java: ########## @@ -0,0 +1,108 @@ +/* + * 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; Review Comment: The field name `additionInfos` is awkward/unclear; `additionalInfos` (or `additionalInfo` if singular) is more standard and improves readability throughout this class. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
