update to AWS SDK v2

Talend · Nov 11, 2024 · c999469 · c999469
1 parent 7615204
commit c999469
Show file tree

Hide file tree

Showing 4 changed files with 30 additions and 14 deletions.
diff --git a/.github/workflows/build_and_test.yml b/.github/workflows/build_and_test.yml
@@ -1055,7 +1055,7 @@ jobs:
           kubectl create clusterrolebinding serviceaccounts-cluster-admin --clusterrole=cluster-admin --group=system:serviceaccounts || true
           kubectl apply -f https://raw.githubusercontent.com/volcano-sh/volcano/v1.7.0/installer/volcano-development.yaml || true
           eval $(minikube docker-env)
-          build/sbt -Psparkr -Pkubernetes -Pvolcano -Pkubernetes-integration-tests -Dspark.kubernetes.test.volcanoMaxConcurrencyJobNum=1 -Dtest.exclude.tags=local "kubernetes-integration-tests/test"
+          build/sbt -Phadoop-3 -Psparkr -Pkubernetes -Pvolcano -Pkubernetes-integration-tests -Dspark.kubernetes.test.volcanoMaxConcurrencyJobNum=1 -Dtest.exclude.tags=local "kubernetes-integration-tests/test"
       - name: Upload Spark on K8S integration tests log files
         if: failure()
         uses: actions/upload-artifact@v3

diff --git a/pom.xml b/pom.xml
@@ -162,6 +162,7 @@
     <aws.kinesis.client.version>1.12.0</aws.kinesis.client.version>
     <!-- Should be consistent with Kinesis client dependency -->
     <aws.java.sdk.version>1.11.655</aws.java.sdk.version>
+    <aws.java.sdk.v2.version>2.20.160</aws.java.sdk.v2.version>
     <!-- the producer is used in tests -->
     <aws.kinesis.producer.version>0.12.8</aws.kinesis.producer.version>
     <gcs-connector.version>hadoop3-2.2.14</gcs-connector.version>

diff --git a/resource-managers/kubernetes/integration-tests/pom.xml b/resource-managers/kubernetes/integration-tests/pom.xml
@@ -210,9 +210,9 @@
       </activation>
       <dependencies>
         <dependency>
-          <groupId>com.amazonaws</groupId>
-          <artifactId>aws-java-sdk-bundle</artifactId>
-          <version>1.11.375</version>
+          <groupId>software.amazon.awssdk</groupId>
+          <artifactId>bundle</artifactId>
+          <version>${aws.java.sdk.v2.version}</version>
           <scope>test</scope>
         </dependency>
       </dependencies>

diff --git a/...ion-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/DepsTestsSuite.scala b/...ion-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/DepsTestsSuite.scala
@@ -17,22 +17,25 @@
 package org.apache.spark.deploy.k8s.integrationtest
 
 import java.io.File
-import java.net.URL
+import java.net.{URI, URL}
 import java.nio.file.Files
 
 import scala.collection.JavaConverters._
 
-import com.amazonaws.auth.BasicAWSCredentials
-import com.amazonaws.services.s3.AmazonS3Client
 import io.fabric8.kubernetes.api.model._
 import io.fabric8.kubernetes.api.model.apps.StatefulSetBuilder
 import org.apache.hadoop.util.VersionInfo
 import org.scalatest.concurrent.{Eventually, PatienceConfiguration}
 import org.scalatest.time.{Minutes, Span}
+import software.amazon.awssdk.auth.credentials.{AwsBasicCredentials, StaticCredentialsProvider}
+import software.amazon.awssdk.core.sync.RequestBody
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.services.s3.S3Client
+import software.amazon.awssdk.services.s3.model.{CreateBucketRequest, PutObjectRequest}
 
 import org.apache.spark.SparkException
 import org.apache.spark.deploy.k8s.integrationtest.DepsTestsSuite.{DEPS_TIMEOUT, FILE_CONTENTS, HOST_PATH}
-import org.apache.spark.deploy.k8s.integrationtest.KubernetesSuite.{INTERVAL, MinikubeTag, SPARK_PI_MAIN_CLASS, TIMEOUT}
+import org.apache.spark.deploy.k8s.integrationtest.KubernetesSuite._
 import org.apache.spark.deploy.k8s.integrationtest.Utils.getExamplesJarName
 import org.apache.spark.deploy.k8s.integrationtest.backend.minikube.Minikube
 import org.apache.spark.internal.config.{ARCHIVES, PYSPARK_DRIVER_PYTHON, PYSPARK_PYTHON}
@@ -45,6 +48,7 @@ private[spark] trait DepsTestsSuite { k8sSuite: KubernetesSuite =>
   val BUCKET = "spark"
   val ACCESS_KEY = "minio"
   val SECRET_KEY = "miniostorage"
+  val REGION = "us-west-2"
 
   private def getMinioContainer(): Container = {
     val envVars = Map (
@@ -302,20 +306,27 @@ private[spark] trait DepsTestsSuite { k8sSuite: KubernetesSuite =>
   private def getS3Client(
       endPoint: String,
       accessKey: String = ACCESS_KEY,
-      secretKey: String = SECRET_KEY): AmazonS3Client = {
-    val credentials = new BasicAWSCredentials(accessKey, secretKey)
-    val s3client = new AmazonS3Client(credentials)
-    s3client.setEndpoint(endPoint)
+      secretKey: String = SECRET_KEY): S3Client = {
+    val credentials = AwsBasicCredentials.create(accessKey, secretKey)
+    val s3client = S3Client.builder()
+      .credentialsProvider(StaticCredentialsProvider.create(credentials))
+      .endpointOverride(URI.create(endPoint))
+      .region(Region.of(REGION))
+      .build()
     s3client
   }
 
   private def createS3Bucket(accessKey: String, secretKey: String, endPoint: String): Unit = {
     Eventually.eventually(TIMEOUT, INTERVAL) {
       try {
         val s3client = getS3Client(endPoint, accessKey, secretKey)
-        s3client.createBucket(BUCKET)
+        val createBucketRequest = CreateBucketRequest.builder()
+          .bucket(BUCKET)
+          .build()
+        s3client.createBucket(createBucketRequest)
       } catch {
         case e: Exception =>
+          logError(s"Failed to create bucket $BUCKET", e)
           throw new SparkException(s"Failed to create bucket $BUCKET.", e)
       }
     }
@@ -328,7 +339,11 @@ private[spark] trait DepsTestsSuite { k8sSuite: KubernetesSuite =>
     Eventually.eventually(TIMEOUT, INTERVAL) {
       try {
         val s3client = getS3Client(endPoint)
-        s3client.putObject(BUCKET, objectKey, objectContent)
+        val putObjectRequest = PutObjectRequest.builder()
+          .bucket(BUCKET)
+          .key(objectKey)
+          .build()
+        s3client.putObject(putObjectRequest, RequestBody.fromString(objectContent))
       } catch {
         case e: Exception =>
           throw new SparkException(s"Failed to create object $BUCKET/$objectKey.", e)