Fixed the issue of the fluss-fs-s3 jar becoming too big due to Hadoop version upgrades.

Signed-off-by: Pei Yu <125331682@qq.com>

Author: sd4324530

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

@@ -233,6 +233,10 @@
                     <groupId>org.slf4j</groupId>
                     <artifactId>slf4j-reload4j</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>software.amazon.awssdk</groupId>
+                    <artifactId>bundle</artifactId>
+                </exclusion>
             </exclusions>
         </dependency>
 

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

@@ -0,0 +1,32 @@
+/*
+ * 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.exception;
+
+/** Exception thrown when no credentials were found. */
+public class InvalidCredentialsException extends RuntimeException {
+
+    public static final String E_NO_AWS_CREDENTIALS = "No AWS Credentials";
+
+    public InvalidCredentialsException(String credentialProvider) {
+        this(credentialProvider, null);
+    }
+
+    public InvalidCredentialsException(String credentialProvider, Throwable throwable) {
+        super(credentialProvider + ": " + E_NO_AWS_CREDENTIALS, throwable);
+    }
+}

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

@@ -17,12 +17,12 @@
 
 package org.apache.fluss.fs.s3.token;
 
+import org.apache.fluss.fs.s3.exception.InvalidCredentialsException;
 import org.apache.fluss.fs.token.Credentials;
 
 import com.amazonaws.auth.AWSCredentials;
 import com.amazonaws.auth.AWSCredentialsProvider;
 import com.amazonaws.auth.BasicSessionCredentials;
-import org.apache.hadoop.fs.s3a.auth.NoAwsCredentialsException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -46,7 +46,7 @@ public AWSCredentials getCredentials() {
         Credentials credentials = S3DelegationTokenReceiver.getCredentials();
 
         if (credentials == null) {
-            throw new NoAwsCredentialsException(COMPONENT);
+            throw new InvalidCredentialsException(COMPONENT);
         }
         LOG.debug("Providing session credentials");
         return new BasicSessionCredentials(

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

@@ -17,12 +17,12 @@
 
 package org.apache.fluss.fs.s3.token;
 
+import org.apache.fluss.fs.s3.exception.InvalidCredentialsException;
 import org.apache.fluss.fs.token.Credentials;
 import org.apache.fluss.fs.token.CredentialsJsonSerde;
 import org.apache.fluss.fs.token.ObtainedSecurityToken;
 import org.apache.fluss.fs.token.SecurityTokenReceiver;
 
-import org.apache.hadoop.fs.s3a.auth.NoAwsCredentialsException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -63,7 +63,7 @@ public static void updateHadoopConfig(org.apache.hadoop.conf.Configuration hadoo
         if (additionInfos == null) {
             // if addition info is null, it also means we have not received any token,
             // we throw InvalidCredentialsException
-            throw new NoAwsCredentialsException(DynamicTemporaryAWSCredentialsProvider.COMPONENT);
+            throw new InvalidCredentialsException(DynamicTemporaryAWSCredentialsProvider.COMPONENT);
         } else {
             for (Map.Entry<String, String> entry : additionInfos.entrySet()) {
                 hadoopConfig.set(entry.getKey(), entry.getValue());