Remove notificationwlambda's dependency on awssdkv1

Author: jorgeazevedo

build.sbt

@@ -122,6 +122,7 @@ lazy val common = project
       "com.gu" %% "mobile-logstash-encoder" % "1.1.8",
       "com.gu" %% "simple-configuration-ssm" % simpleConfigurationVersion,
       "software.amazon.awssdk" % "regions" % "2.25.13",
+      "software.amazon.awssdk" % "s3" % "2.29.24",
       "org.postgresql" % "postgresql" % "42.7.7",
       "ch.qos.logback" % "logback-core" % logbackVersion,
       "ch.qos.logback" % "logback-classic" % logbackVersion,
@@ -423,9 +424,9 @@ lazy val notificationworkerlambda = lambda("notificationworkerlambda", "notifica
       "com.google.protobuf" % "protobuf-java" % "4.31.1",
       "com.google.protobuf" % "protobuf-java-util" % "4.31.1",
       "com.amazonaws" % "aws-lambda-java-events" % "2.2.9",
-      "com.amazonaws" % "aws-java-sdk-sqs" % awsSdkVersion,
-      "com.amazonaws" % "aws-java-sdk-s3" % awsSdkVersion,
-      "com.amazonaws" % "amazon-sqs-java-messaging-lib" % "2.1.4",
+      "software.amazon.awssdk" % "s3" % "2.29.24",
+      "software.amazon.awssdk" % "sqs" % "2.29.24",
+      "software.amazon.awssdk" % "cloudwatch" % "2.29.24",
       "com.squareup.okhttp3" % "okhttp" % okHttpVersion,
       "org.playframework" %% "play-json" % playJsonVersion,
       "com.google.oauth-client" % "google-oauth-client" % googleOAuthClient,
@@ -450,6 +451,15 @@ lazy val notificationworkerlambda = lambda("notificationworkerlambda", "notifica
       // gRPC modules not needed for FCM HTTP API
       ExclusionRule("io.grpc", "grpc-xds"), // 34.24MB
       ExclusionRule("io.grpc", "grpc-netty"), // 9.28MB
+      
+      // Exclude AWS SDK v1 except lambda-java-events which is needed for Lambda runtime
+      ExclusionRule("com.amazonaws", "aws-java-sdk-s3"),
+      ExclusionRule("com.amazonaws", "aws-java-sdk-sqs"),
+      ExclusionRule("com.amazonaws", "aws-java-sdk-cloudwatch"),
+      ExclusionRule("com.amazonaws", "aws-java-sdk-dynamodb"),
+      ExclusionRule("com.amazonaws", "aws-java-sdk-athena"),
+      ExclusionRule("com.amazonaws", "aws-java-sdk-core"),
+      ExclusionRule("com.amazonaws", "amazon-sqs-java-messaging-lib"),
     ),
 )
 

common/src/main/scala/aws/S3.scala

@@ -1,10 +1,8 @@
 package aws
 
-import java.io.{ByteArrayInputStream, InputStream}
-
-import com.amazonaws.services.s3.AmazonS3
-import com.amazonaws.services.s3.model.{ObjectMetadata, PutObjectRequest, PutObjectResult, S3Object}
-import com.amazonaws.util.IOUtils
+import software.amazon.awssdk.services.s3.S3Client
+import software.amazon.awssdk.services.s3.model.{GetObjectRequest, PutObjectRequest, PutObjectResponse}
+import software.amazon.awssdk.core.sync.RequestBody
 import exception.TopicCounterException
 import models.TopicCount
 import org.slf4j.{Logger, LoggerFactory}
@@ -15,18 +13,19 @@ import scala.util.{Failure, Success, Try}
 
 trait S3[T]  {
 
-  def s3Client: AmazonS3
+  def s3Client: S3Client
   def bucketName: String
   def path: String
   def logger: Logger
 
-  def put(data: Seq[T])(implicit format: Format[T]) : PutObjectResult =  {
-    val (inputStream, contentLength) = jsonInputStreamAndContentLength(data)
-    val metaData: ObjectMetadata = new ObjectMetadata()
-    metaData.setContentType("application/json")
-    metaData.setContentLength(contentLength)
-    val putObjectRequest: PutObjectRequest = new PutObjectRequest(bucketName, path, inputStream, metaData)
-    s3Client.putObject(putObjectRequest)
+  def put(data: Seq[T])(implicit format: Format[T]) : PutObjectResponse =  {
+    val jsonAsBytes = Json.toBytes(Json.toJson(data))
+    val putObjectRequest = PutObjectRequest.builder()
+      .bucket(bucketName)
+      .key(path)
+      .contentType("application/json")
+      .build()
+    s3Client.putObject(putObjectRequest, RequestBody.fromBytes(jsonAsBytes))
   }
 
   def fetch()(implicit format: Format[T], executionException: ExecutionContext) : Future[List[T]] = {
@@ -39,7 +38,7 @@ trait S3[T]  {
   }
 
   private def parseS3Object()(implicit format: Format[T]) : List[T] = {
-    Json.fromJson[List[T]](Json.parse(asString(s3Client.getObject(bucketName, path)))) match {
+    Json.fromJson[List[T]](Json.parse(asString())) match {
       case JsSuccess(list, __) =>
         logger.debug(s"Got ${list.length} topic counts from s3")
         list
@@ -50,23 +49,17 @@ trait S3[T]  {
     }
   }
 
-  private def asString(s3Object: S3Object): String = {
-     val s3ObjectContent = s3Object.getObjectContent
-     try {
-       IOUtils.toString(s3ObjectContent)
-     }
-     finally {
-       s3ObjectContent.close()
-     }
-  }
-
-  private def jsonInputStreamAndContentLength(data: Seq[T])(implicit format: Format[T]): (InputStream, Long) = {
-    val jsonAsBytes  = Json.toBytes(Json.toJson(data))
-    (new ByteArrayInputStream(jsonAsBytes), jsonAsBytes.length)
+  private def asString(): String = {
+    val getObjectRequest = GetObjectRequest.builder()
+      .bucket(bucketName)
+      .key(path)
+      .build()
+    val response = s3Client.getObjectAsBytes(getObjectRequest)
+    response.asUtf8String()
   }
 }
 
-class TopicCountsS3(override val s3Client: AmazonS3, override val bucketName: String, override val path: String) extends S3[TopicCount]  {
+class TopicCountsS3(override val s3Client: S3Client, override val bucketName: String, override val path: String) extends S3[TopicCount]  {
   override def logger: Logger = LoggerFactory.getLogger(this.getClass)
 }
 

notificationworkerlambda/src/main/scala/com/gu/notifications/worker/TopicCounterLambda.scala

@@ -2,9 +2,9 @@ package com.gu.notifications.worker
 
 import aws.TopicCountsS3
 import cats.effect.{ContextShift, IO}
-import com.amazonaws.auth.AWSCredentialsProviderChain
-import com.amazonaws.regions.Regions
-import com.amazonaws.services.s3.{AmazonS3, AmazonS3ClientBuilder}
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.services.s3.S3Client
 import com.gu.notifications.worker.utils.{Aws, Logging}
 import db.{DatabaseConfig, RegistrationService}
 import doobie.util.transactor.Transactor
@@ -17,11 +17,11 @@ class TopicCounterLambda extends Logging {
 
   def env = Env()
 
-  lazy val credentials: AWSCredentialsProviderChain = Aws.credentialsProvider
+  lazy val credentials: AwsCredentialsProvider = Aws.credentialsProviderV2
 
-  lazy val s3Client: AmazonS3 = AmazonS3ClientBuilder.standard
-    .withRegion(Regions.EU_WEST_1)
-    .withCredentials(credentials)
+  lazy val s3Client: S3Client = S3Client.builder()
+    .region(Region.EU_WEST_1)
+    .credentialsProvider(credentials)
     .build()
 
   implicit val ec: ExecutionContextExecutor = ExecutionContext.global
@@ -41,6 +41,6 @@ class TopicCounterLambda extends Logging {
 
   def runLocally(): Unit = {
     topicCounts.handleRequest()
-    s3Client.shutdown()
+    s3Client.close()
   }
 }

notificationworkerlambda/src/main/scala/com/gu/notifications/worker/cleaning/CleaningClient.scala

@@ -1,8 +1,9 @@
 package com.gu.notifications.worker.cleaning
 
 import cats.effect.IO
-import com.amazonaws.regions.Regions
-import com.amazonaws.services.sqs.{AmazonSQS, AmazonSQSClient}
+import software.amazon.awssdk.services.sqs.SqsClient
+import software.amazon.awssdk.services.sqs.model.SendMessageRequest
+import software.amazon.awssdk.regions.Region
 import com.gu.notifications.worker.models.InvalidTokens
 import com.gu.notifications.worker.utils.Aws
 import fs2.{Chunk, Pipe}
@@ -15,15 +16,17 @@ trait CleaningClient {
 
 class CleaningClientImpl(sqsUrl: String) extends CleaningClient {
 
-  val sqsClient: AmazonSQS = AmazonSQSClient
-    .builder()
-    .withCredentials(Aws.credentialsProvider)
-    .withRegion(Regions.EU_WEST_1)
-    .build
+  val sqsClient: SqsClient = SqsClient.builder()
+    .region(Region.EU_WEST_1)
+    .credentialsProvider(Aws.credentialsProviderV2)
+    .build()
 
   private def sendTokensToQueue(tokens: List[String])(implicit logger: Logger): Unit = {
     val json = Json.stringify(Json.toJson(InvalidTokens(tokens)))
-    sqsClient.sendMessage(sqsUrl, json)
+    sqsClient.sendMessage(SendMessageRequest.builder()
+      .queueUrl(sqsUrl)
+      .messageBody(json)
+      .build())
     logger.info(s"Sent ${tokens.size} tokens for deletion via SQS")
   }
 

notificationworkerlambda/src/main/scala/com/gu/notifications/worker/tokens/SqsDeliveryService.scala

@@ -3,10 +3,9 @@ package com.gu.notifications.worker.tokens
 import java.util.concurrent.TimeUnit
 
 import cats.effect._
-import com.amazonaws.handlers.AsyncHandler
-import com.amazonaws.regions.Regions
-import com.amazonaws.services.sqs.AmazonSQSAsyncClientBuilder
-import com.amazonaws.services.sqs.model.{SendMessageRequest, SendMessageResult}
+import software.amazon.awssdk.services.sqs.SqsAsyncClient
+import software.amazon.awssdk.services.sqs.model.SendMessageRequest
+import software.amazon.awssdk.regions.Region
 import com.gu.notifications.worker.utils.Aws
 import fs2.Stream
 import play.api.libs.json.Json
@@ -24,24 +23,25 @@ class SqsDeliveryServiceImpl[F[_]](queueUrl: String)(implicit ece: ExecutionCont
   F: Async[F],
   T: Timer[F]
 ) extends SqsDeliveryService[F] {
-  val amazonSQSAsyncClient = AmazonSQSAsyncClientBuilder.standard()
-    .withCredentials(Aws.credentialsProvider)
-    .withRegion(Regions.EU_WEST_1)
+  val sqsAsyncClient = SqsAsyncClient.builder()
+    .region(Region.EU_WEST_1)
+    .credentialsProvider(Aws.credentialsProviderV2)
     .build()
 
   def send(chunkedTokensBatch: ChunkedTokens)(oncomplete: Either[Throwable, Unit] => Unit): Unit = {
-    val sendMessageRequest = new SendMessageRequest()
-      .withMessageBody(Json.stringify(Json.toJson(chunkedTokensBatch)))
-      .withQueueUrl(queueUrl)
+    val sendMessageRequest = SendMessageRequest.builder()
+      .messageBody(Json.stringify(Json.toJson(chunkedTokensBatch)))
+      .queueUrl(queueUrl)
+      .build()
 
-    val handler = new AsyncHandler[SendMessageRequest, SendMessageResult] {
-      override def onError(exception: Exception): Unit = oncomplete(Left(new Exception(Json.stringify(Json.toJson(chunkedTokensBatch)), exception)))
-
-      override def onSuccess(request: SendMessageRequest, result: SendMessageResult): Unit = oncomplete(Right(()))
-    }
-    amazonSQSAsyncClient.sendMessageAsync(
-      sendMessageRequest,
-      handler)
+    sqsAsyncClient.sendMessage(sendMessageRequest)
+      .whenComplete { (_, error) =>
+        if (error != null) {
+          oncomplete(Left(new Exception(Json.stringify(Json.toJson(chunkedTokensBatch)), error)))
+        } else {
+          oncomplete(Right(()))
+        }
+      }
   }
 
 

notificationworkerlambda/src/main/scala/com/gu/notifications/worker/utils/Aws.scala

@@ -3,5 +3,5 @@ package com.gu.notifications.worker.utils
 import utils.MobileAwsCredentialsProvider
 
 object Aws {
-  lazy val credentialsProvider = new MobileAwsCredentialsProvider
+  lazy val credentialsProviderV2 = MobileAwsCredentialsProvider.mobileAwsCredentialsProviderv2
 }