address copilot's comments

Author: fresh-borzoni

fluss-filesystems/fluss-fs-s3/src/main/java/org/apache/fluss/fs/s3/S3FileSystemPlugin.java

@@ -17,6 +17,7 @@
 
 package org.apache.fluss.fs.s3;
 
+import org.apache.fluss.annotation.VisibleForTesting;
 import org.apache.fluss.config.ConfigBuilder;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.fs.FileSystem;
@@ -44,6 +45,7 @@ public class S3FileSystemPlugin implements FileSystemPlugin {
     private static final String HADOOP_CONFIG_PREFIX = "fs.s3a.";
 
     private static final String ACCESS_KEY_ID = "fs.s3a.access.key";
+    private static final String ACCESS_KEY_SECRET = "fs.s3a.secret.key";
 
     private static final String ROLE_ARN_KEY = "fs.s3a.assumed.role.arn";
 
@@ -123,8 +125,11 @@ private URI getInitURI(URI fsUri, org.apache.hadoop.conf.Configuration hadoopCon
         return fsUri;
     }
 
-    private void setCredentialProvider(org.apache.hadoop.conf.Configuration hadoopConfig) {
-        boolean hasStaticKeys = hadoopConfig.get(ACCESS_KEY_ID) != null;
+    @VisibleForTesting
+    void setCredentialProvider(org.apache.hadoop.conf.Configuration hadoopConfig) {
+        boolean hasStaticKeys =
+                hadoopConfig.get(ACCESS_KEY_ID) != null
+                        && hadoopConfig.get(ACCESS_KEY_SECRET) != null;
         boolean hasRoleArn = hadoopConfig.get(ROLE_ARN_KEY) != null;
 
         if (hasStaticKeys || hasRoleArn) {

fluss-filesystems/fluss-fs-s3/src/main/java/org/apache/fluss/fs/s3/token/S3DelegationTokenProvider.java

@@ -40,6 +40,7 @@
 import java.util.Map;
 import java.util.UUID;
 
+import static org.apache.fluss.utils.Preconditions.checkArgument;
 import static org.apache.fluss.utils.Preconditions.checkNotNull;
 
 /** Delegation token provider for S3 Hadoop filesystems. */
@@ -73,9 +74,12 @@ public S3DelegationTokenProvider(String scheme, Configuration conf) {
         this.roleArn = conf.get(ROLE_ARN_KEY);
         this.stsEndpoint = conf.get(STS_ENDPOINT_KEY);
 
-        if (accessKey == null || secretKey == null) {
-            checkNotNull(
-                    roleArn,
+        checkArgument(
+                (accessKey == null) == (secretKey == null),
+                "fs.s3a.access.key and fs.s3a.secret.key must both be set or both be unset.");
+        if (accessKey == null) {
+            checkArgument(
+                    roleArn != null,
                     ROLE_ARN_KEY + " must be set when static credentials are not provided.");
         }
 

fluss-filesystems/fluss-fs-s3/src/test/java/org/apache/fluss/fs/s3/S3FileSystemPluginTest.java

@@ -0,0 +1,84 @@
+/*
+ * 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.s3;
+
+import org.apache.fluss.fs.s3.token.DynamicTemporaryAWSCredentialsProvider;
+import org.apache.fluss.fs.s3.token.S3DelegationTokenReceiver;
+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 org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for server/client detection in {@link S3FileSystemPlugin}. */
+class S3FileSystemPluginTest {
+
+    private static final String PROVIDER_CONFIG = "fs.s3a.aws.credentials.provider";
+
+    @Test
+    void testServerModeWithStaticKeys() {
+        Configuration hadoopConfig = new Configuration();
+        hadoopConfig.set("fs.s3a.access.key", "testAccessKey");
+        hadoopConfig.set("fs.s3a.secret.key", "testSecretKey");
+
+        S3FileSystemPlugin plugin = new S3FileSystemPlugin();
+        plugin.setCredentialProvider(hadoopConfig);
+
+        String providers = hadoopConfig.get(PROVIDER_CONFIG, "");
+        assertThat(providers).doesNotContain(DynamicTemporaryAWSCredentialsProvider.NAME);
+    }
+
+    @Test
+    void testServerModeWithRoleArnOnly() {
+        Configuration hadoopConfig = new Configuration();
+        hadoopConfig.set("fs.s3a.assumed.role.arn", "arn:aws:iam::123456789012:role/test-role");
+
+        S3FileSystemPlugin plugin = new S3FileSystemPlugin();
+        plugin.setCredentialProvider(hadoopConfig);
+
+        String providers = hadoopConfig.get(PROVIDER_CONFIG, "");
+        assertThat(providers).doesNotContain(DynamicTemporaryAWSCredentialsProvider.NAME);
+    }
+
+    @Test
+    void testClientModeWithDelegatedCredentials() {
+        // Pre-populate receiver so updateHadoopConfig does not throw.
+        Credentials creds = new Credentials("testKey", "testSecret", "testToken");
+        ObtainedSecurityToken token =
+                new ObtainedSecurityToken(
+                        "s3",
+                        CredentialsJsonSerde.toJson(creds),
+                        System.currentTimeMillis() + 3600000,
+                        Collections.singletonMap("fs.s3a.region", "us-east-1"));
+        S3DelegationTokenReceiver receiver = new S3DelegationTokenReceiver();
+        receiver.onNewTokensObtained(token);
+
+        Configuration hadoopConfig = new Configuration();
+
+        S3FileSystemPlugin plugin = new S3FileSystemPlugin();
+        plugin.setCredentialProvider(hadoopConfig);
+
+        String providers = hadoopConfig.get(PROVIDER_CONFIG, "");
+        assertThat(providers).contains(DynamicTemporaryAWSCredentialsProvider.NAME);
+    }
+}

fluss-filesystems/fluss-fs-s3/src/test/java/org/apache/fluss/fs/s3/token/S3DelegationTokenProviderTest.java

@@ -41,7 +41,7 @@ void testDefaultChainWithoutRoleArnThrows() {
         conf.set("fs.s3a.region", "us-east-1");
 
         assertThatThrownBy(() -> new S3DelegationTokenProvider("s3", conf))
-                .isInstanceOf(NullPointerException.class)
+                .isInstanceOf(IllegalArgumentException.class)
                 .hasMessageContaining("fs.s3a.assumed.role.arn must be set");
     }
 }