Merge pull request #1426 from guardian/ld/java-sdk-upgrade-kinesis

Java sdk upgrade kinesis: Update path (7)

Author: lindseydew

app/data/DataStores.scala

@@ -31,15 +31,15 @@ class DataStores(aws: AWSConfig with SNSAccess, capi: CapiAccess) {
       new NotifyingAtomPublisher(
         isLive = true,
         topicArn = capiContentEventsTopicName,
-        underlying = new LiveKinesisAtomPublisher(
+        underlying = new LiveKinesisAtomPublisherV2(
           aws.liveKinesisStreamName,
           aws.crossAccountKinesisClient
         ),
         sns = aws.snsClient
       )
 
     case None =>
-      new LiveKinesisAtomPublisher(
+      new LiveKinesisAtomPublisherV2(
         aws.liveKinesisStreamName,
         aws.crossAccountKinesisClient
       )
@@ -50,28 +50,28 @@ class DataStores(aws: AWSConfig with SNSAccess, capi: CapiAccess) {
       new NotifyingAtomPublisher(
         isLive = true,
         topicArn = capiContentEventsTopicName,
-        underlying = new PreviewKinesisAtomPublisher(
+        underlying = new PreviewKinesisAtomPublisherV2(
           aws.previewKinesisStreamName,
           aws.crossAccountKinesisClient
         ),
         sns = aws.snsClient
       )
 
     case None =>
-      new PreviewKinesisAtomPublisher(
+      new PreviewKinesisAtomPublisherV2(
         aws.previewKinesisStreamName,
         aws.crossAccountKinesisClient
       )
   }
 
   val reindexPreview: PreviewAtomReindexer =
-    new PreviewKinesisAtomReindexer(
+    new PreviewKinesisAtomReindexerV2(
       aws.previewKinesisReindexStreamName,
       aws.crossAccountKinesisClient
     )
 
-  val reindexPublished: PublishedKinesisAtomReindexer =
-    new PublishedKinesisAtomReindexer(
+  val reindexPublished: PublishedKinesisAtomReindexerV2 =
+    new PublishedKinesisAtomReindexerV2(
       aws.publishedKinesisReindexStreamName,
       aws.crossAccountKinesisClient
     )

build.sbt

@@ -8,7 +8,7 @@ val scroogeVersion = "4.12.0"
 val awsVersion = "1.11.1034"
 val awsV2Version = "2.32.26"
 val pandaVersion = "10.0.0"
-val atomMakerVersion = "6.0.0"
+val atomMakerVersion = "8.0.0"
 val typesafeConfigVersion =
   "1.4.0" // to match what we get from Play transitively
 val scanamoVersion = "1.0.0-M28"
@@ -104,7 +104,7 @@ lazy val common = (project in file("common"))
       "com.typesafe" % "config" % typesafeConfigVersion,
       "com.amazonaws" % "aws-lambda-java-core" % awsLambdaCoreVersion,
       "software.amazon.awssdk" % "dynamodb" % awsV2Version,
-      "com.amazonaws" % "aws-java-sdk-kinesis" % awsVersion,
+      "software.amazon.awssdk" % "kinesis" % awsV2Version,
       "com.gu" %% "play-json-extensions" % playJsonExtensionsVersion,
       "ch.qos.logback" % "logback-classic" % logbackClassicVersion,
       "com.amazonaws" % "aws-java-sdk-sts" % awsVersion,

common/src/main/scala/com/gu/media/aws/KinesisAccess.scala

@@ -1,13 +1,15 @@
 package com.gu.media.aws
 
-import java.nio.ByteBuffer
-import java.nio.charset.StandardCharsets
-
-import com.amazonaws.AmazonClientException
-import com.amazonaws.services.kinesis.AmazonKinesisClientBuilder
+import software.amazon.awssdk.core.exception.SdkClientException
+import software.amazon.awssdk.services.kinesis.model.{
+  PutRecordRequest,
+  DescribeStreamRequest
+}
+import software.amazon.awssdk.services.kinesis.KinesisClient
 import com.gu.media.Settings
 import com.gu.media.logging.Logging
 import play.api.libs.json.{Json, Writes}
+import software.amazon.awssdk.core.SdkBytes
 
 trait KinesisAccess { this: Settings with AwsAccess with Logging =>
   val liveKinesisStreamName: String = getMandatoryString(
@@ -30,16 +32,16 @@ trait KinesisAccess { this: Settings with AwsAccess with Logging =>
 
   val syncWithPluto: Boolean = getBoolean("pluto.sync").getOrElse(false)
 
-  lazy val crossAccountKinesisClient = AmazonKinesisClientBuilder
-    .standard()
-    .withCredentials(credentials.crossAccount.awsV1Creds)
-    .withRegion(region.getName)
+  lazy val crossAccountKinesisClient = KinesisClient
+    .builder()
+    .credentialsProvider(credentials.crossAccount.awsV2Creds)
+    .region(awsV2Region)
     .build()
 
-  lazy val kinesisClient = AmazonKinesisClientBuilder
-    .standard()
-    .withCredentials(credentials.instance.awsV1Creds)
-    .withRegion(region.getName)
+  lazy val kinesisClient = KinesisClient
+    .builder()
+    .credentialsProvider(credentials.instance.awsV2Creds)
+    .region(awsV2Region)
     .build()
 
   def sendOnKinesis[T: Writes](
@@ -50,16 +52,23 @@ trait KinesisAccess { this: Settings with AwsAccess with Logging =>
     val json = Json.stringify(Json.toJson(value))
     log.info(s"Sending JSON on Kinesis [$streamName]: $json")
 
-    val bytes = ByteBuffer.wrap(json.getBytes(StandardCharsets.UTF_8))
+    val putRecordRequest = PutRecordRequest
+      .builder()
+      .streamName(streamName)
+      .data(SdkBytes.fromUtf8String(json))
+      .partitionKey(partitionKey)
+      .build()
 
-    kinesisClient.putRecord(streamName, bytes, partitionKey)
+    kinesisClient.putRecord(putRecordRequest)
   }
 
   def testKinesisAccess(streamName: String): Boolean = try {
-    crossAccountKinesisClient.describeStream(streamName)
+    val describeStream =
+      DescribeStreamRequest.builder().streamName(streamName).build()
+    crossAccountKinesisClient.describeStream(describeStream)
     true
   } catch {
-    case e: AmazonClientException =>
+    case e: SdkClientException =>
       false
   }
 }

uploader/src/main/scala/com/gu/media/upload/AddAssetToAtom.scala

@@ -2,7 +2,10 @@ package com.gu.media.upload
 
 import java.util.Date
 import com.gu.atom.data.PreviewDynamoDataStore
-import com.gu.atom.publish.PreviewKinesisAtomPublisher
+import com.gu.atom.publish.{
+  PreviewKinesisAtomPublisher,
+  PreviewKinesisAtomPublisherV2
+}
 import com.gu.contentatom.thrift.{Atom, ContentAtomEvent, EventType}
 import com.gu.media.aws.{DynamoAccess, KinesisAccess, UploadAccess}
 import com.gu.media.lambda.LambdaWithParams
@@ -30,7 +33,7 @@ class AddAssetToAtom
   )
 
   private val store = new PreviewDynamoDataStore(dynamoDB, dynamoTableName)
-  private val publisher = new PreviewKinesisAtomPublisher(
+  private val publisher = new PreviewKinesisAtomPublisherV2(
     previewKinesisStreamName,
     crossAccountKinesisClient
   )