Skip to content

Commit c48fa19

Browse files
oermolaevoermolaev
authored
Merge branch 'main' into updates/kinesis-client
2 parents 22e4776 + 1962d43 commit c48fa19

10 files changed

Lines changed: 184 additions & 185 deletions

File tree

‎modules/cloudutils/aws/src/main/scala/com/snowplowanalytics/snowplow/enrich/aws/S3Client.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,7 @@ object S3Client {
5858
}
5959

6060
private def getRegion(): Region =
61-
Either.catchNonFatal((new DefaultAwsRegionProviderChain).getRegion) match {
61+
Either.catchNonFatal(DefaultAwsRegionProviderChain.builder().build().getRegion) match {
6262
case Right(region) =>
6363
region
6464
case _ =>

‎modules/eventbridge/src/it/scala/com/snowplowanalytics/snowplow/enrich/kinesis/Sink.scala‎

Lines changed: 43 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -12,10 +12,10 @@
1212
*/
1313
package com.snowplowanalytics.snowplow.enrich.kinesis
1414

15-
import java.nio.ByteBuffer
15+
import java.nio.charset.StandardCharsets
1616
import java.util.UUID
1717

18-
import scala.collection.JavaConverters._
18+
import scala.jdk.CollectionConverters._
1919

2020
import cats.implicits._
2121
import cats.{Monoid, Parallel}
@@ -29,10 +29,10 @@ import org.typelevel.log4cats.slf4j.Slf4jLogger
2929
import retry.syntax.all._
3030
import retry.RetryPolicy
3131

32-
import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration
33-
34-
import com.amazonaws.services.kinesis.model._
35-
import com.amazonaws.services.kinesis.{AmazonKinesis, AmazonKinesisClientBuilder}
32+
import software.amazon.awssdk.core.SdkBytes
33+
import software.amazon.awssdk.regions.Region
34+
import software.amazon.awssdk.services.kinesis.KinesisClient
35+
import software.amazon.awssdk.services.kinesis.model._
3636

3737
import com.snowplowanalytics.snowplow.enrich.common.fs2.{AttributedByteSink, AttributedData, ByteSink}
3838
import com.snowplowanalytics.snowplow.enrich.common.fs2.config.io.Output
@@ -58,8 +58,8 @@ object Sink {
5858
o.region.orElse(getRuntimeRegion) match {
5959
case Some(region) =>
6060
for {
61-
producer <- Resource.eval[F, AmazonKinesis](mkProducer(o, region))
62-
} yield records => writeToKinesis(o, producer, toKinesisRecords(records))
61+
producer <- mkProducer(o, region)
62+
} yield (records: List[AttributedData[Array[Byte]]]) => writeToKinesis(o, producer, toKinesisRecords(records))
6363
case None =>
6464
Resource.eval(Sync[F].raiseError(new RuntimeException(s"Region not found in the config and in the runtime")))
6565
}
@@ -70,23 +70,23 @@ object Sink {
7070
private def mkProducer[F[_]: Sync](
7171
config: Output.Kinesis,
7272
region: String
73-
): F[AmazonKinesis] =
74-
for {
75-
builder <- Sync[F].delay(AmazonKinesisClientBuilder.standard)
76-
withEndpoint <- config.customEndpoint match {
77-
case Some(endpoint) =>
78-
Sync[F].delay(builder.withEndpointConfiguration(new EndpointConfiguration(endpoint.toString, region)))
79-
case None =>
80-
Sync[F].delay(builder.withRegion(region))
81-
}
82-
kinesis <- Sync[F].delay(withEndpoint.build())
83-
_ <- streamExists(kinesis, config.streamName)
84-
} yield kinesis
73+
): Resource[F, KinesisClient] =
74+
Resource
75+
.fromAutoCloseable(Sync[F].delay {
76+
val builder = KinesisClient
77+
.builder()
78+
.region(Region.of(region))
79+
config.customEndpoint
80+
.map(builder.endpointOverride)
81+
.getOrElse(builder)
82+
.build()
83+
})
84+
.evalTap(kinesis => streamExists(kinesis, config.streamName))
8585

86-
private def streamExists[F[_]: Sync](kinesis: AmazonKinesis, stream: String): F[Unit] =
86+
private def streamExists[F[_]: Sync](kinesis: KinesisClient, stream: String): F[Unit] =
8787
for {
88-
described <- Sync[F].delay(kinesis.describeStream(stream))
89-
status = described.getStreamDescription.getStreamStatus
88+
described <- Sync[F].delay(kinesis.describeStream(DescribeStreamRequest.builder().streamName(stream).build()))
89+
status = described.streamDescription.streamStatus.toString
9090
exists <- status match {
9191
case "ACTIVE" | "UPDATING" =>
9292
Sync[F].unit
@@ -97,7 +97,7 @@ object Sink {
9797

9898
private def writeToKinesis[F[_]: Async: Parallel](
9999
config: Output.Kinesis,
100-
kinesis: AmazonKinesis,
100+
kinesis: KinesisClient,
101101
records: List[PutRecordsRequestEntry]
102102
): F[Unit] = {
103103
val policyForErrors = Retries.fullJitter[F](config.backoffPolicy)
@@ -162,7 +162,7 @@ object Sink {
162162
}
163163

164164
private def getRecordSize(record: PutRecordsRequestEntry) =
165-
record.getData.array.size + record.getPartitionKey.getBytes.size
165+
record.data().asByteArray().length + record.partitionKey().getBytes(StandardCharsets.UTF_8).length
166166

167167
/**
168168
* Try writing a batch, and returns a list of the failures to be retried:
@@ -173,7 +173,7 @@ object Sink {
173173
*/
174174
private def tryWriteToKinesis[F[_]: Async](
175175
config: Output.Kinesis,
176-
kinesis: AmazonKinesis,
176+
kinesis: KinesisClient,
177177
records: List[PutRecordsRequestEntry],
178178
retryPolicy: RetryPolicy[F]
179179
): F[Vector[PutRecordsRequestEntry]] =
@@ -203,12 +203,11 @@ object Sink {
203203

204204
private def toKinesisRecords(records: List[AttributedData[Array[Byte]]]): List[PutRecordsRequestEntry] =
205205
records.map { r =>
206-
val binaryData = r.data
207-
val data = ByteBuffer.wrap(binaryData)
208-
val prre = new PutRecordsRequestEntry()
209-
prre.setPartitionKey(r.partitionKey)
210-
prre.setData(data)
211-
prre
206+
PutRecordsRequestEntry
207+
.builder()
208+
.partitionKey(r.partitionKey)
209+
.data(SdkBytes.fromByteArray(r.data))
210+
.build()
212211
}
213212

214213
/**
@@ -244,35 +243,34 @@ object Sink {
244243
)
245244
}
246245

247-
def build(records: List[PutRecordsRequestEntry], prr: PutRecordsResult): TryBatchResult =
248-
if (prr.getFailedRecordCount.toInt =!= 0)
246+
def build(records: List[PutRecordsRequestEntry], prr: PutRecordsResponse): TryBatchResult =
247+
if (prr.failedRecordCount().toInt =!= 0)
249248
records
250-
.zip(prr.getRecords.asScala)
249+
.zip(prr.records().asScala)
251250
.foldMap { case (orig, recordResult) =>
252-
Option(recordResult.getErrorCode) match {
251+
Option(recordResult.errorCode()) match {
253252
case None =>
254253
TryBatchResult(Vector.empty, true, false, None)
255254
case Some("ProvisionedThroughputExceededException") =>
256255
TryBatchResult(Vector(orig), false, true, None)
257256
case Some(_) =>
258-
TryBatchResult(Vector(orig), false, false, Option(recordResult.getErrorMessage))
257+
TryBatchResult(Vector(orig), false, false, Option(recordResult.errorMessage()))
259258
}
260259
}
261260
else
262261
TryBatchResult(Vector.empty, true, false, None)
263262
}
264263

265264
private def putRecords(
266-
kinesis: AmazonKinesis,
265+
kinesis: KinesisClient,
267266
streamName: String,
268267
records: List[PutRecordsRequestEntry]
269-
): PutRecordsResult = {
270-
val putRecordsRequest = {
271-
val prr = new PutRecordsRequest()
272-
prr.setStreamName(streamName)
273-
prr.setRecords(records.asJava)
274-
prr
275-
}
268+
): PutRecordsResponse = {
269+
val putRecordsRequest = PutRecordsRequest
270+
.builder()
271+
.streamName(streamName)
272+
.records(records.asJava)
273+
.build()
276274
kinesis.putRecords(putRecordsRequest)
277275
}
278276

‎modules/eventbridge/src/it/scala/com/snowplowanalytics/snowplow/enrich/kinesis/package.scala‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,9 +12,9 @@
1212
*/
1313
package com.snowplowanalytics.snowplow.enrich
1414

15-
import com.amazonaws.regions.DefaultAwsRegionProviderChain
15+
import software.amazon.awssdk.regions.providers.DefaultAwsRegionProviderChain
1616

1717
package object kinesis {
1818
def getRuntimeRegion: Option[String] =
19-
Option((new DefaultAwsRegionProviderChain).getRegion)
19+
Option(DefaultAwsRegionProviderChain.builder().build().getRegion).map(_.id())
2020
}

‎modules/eventbridge/src/main/scala/com/snowplowanalytics/snowplow/enrich/eventbridge/DynamoDbConfig.scala‎

Lines changed: 32 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -25,14 +25,14 @@ import com.typesafe.config.{ConfigFactory, ConfigValueFactory}
2525

2626
import retry.{RetryDetails, RetryPolicies, RetryPolicy, retryingOnSomeErrors}
2727

28-
import scala.collection.JavaConverters._
28+
import scala.jdk.CollectionConverters._
2929
import scala.concurrent.duration.DurationLong
3030
import scala.util.control.NonFatal
3131

32-
import com.amazonaws.AmazonClientException
33-
import com.amazonaws.services.dynamodbv2.model.ScanRequest
34-
import com.amazonaws.services.dynamodbv2.document.DynamoDB
35-
import com.amazonaws.services.dynamodbv2.{AmazonDynamoDB, AmazonDynamoDBClientBuilder}
32+
import software.amazon.awssdk.core.exception.SdkException
33+
import software.amazon.awssdk.regions.Region
34+
import software.amazon.awssdk.services.dynamodb.DynamoDbClient
35+
import software.amazon.awssdk.services.dynamodb.model.{AttributeValue, GetItemRequest, ScanRequest}
3636

3737
import com.snowplowanalytics.iglu.core.{SchemaKey, SchemaVer}
3838

@@ -80,16 +80,19 @@ object DynamoDbConfig {
8080
region: String,
8181
table: String,
8282
key: String
83-
): F[Base64Hocon] = {
84-
val dynamoDBResource = for {
85-
client <- mkClient[F](region)
86-
api <- Resource.make(Sync[F].delay(new DynamoDB(client)))(a => Sync[F].delay(a.shutdown()))
87-
} yield api
88-
dynamoDBResource.use { dynamoDB =>
83+
): F[Base64Hocon] =
84+
mkClient[F](region).use { dynamoDBClient =>
8985
for {
9086
_ <- unsafeLogger.info(s"Retrieving resolver in DynamoDB $region/$table/$key")
91-
item <- withRetry(Async[F].blocking(dynamoDB.getTable(table).getItem("id", key)))
92-
jsonStr <- Option(item).flatMap(i => Option(i.getString("json"))) match {
87+
getItemRequest = GetItemRequest
88+
.builder()
89+
.tableName(table)
90+
.key(Map("id" -> AttributeValue.builder().s(key).build()).asJava)
91+
.build()
92+
response <- withRetry(Async[F].blocking(dynamoDBClient.getItem(getItemRequest)))
93+
jsonStr <- Option(response.item())
94+
.flatMap(item => Option(item.get("json")))
95+
.flatMap(av => Option(av.s())) match {
9396
case Some(content) =>
9497
Sync[F].pure(content)
9598
case None =>
@@ -100,23 +103,25 @@ object DynamoDbConfig {
100103
}
101104
} yield Base64Hocon(tsConfig)
102105
}
103-
}
104106

105107
private def getEnrichments[F[_]: Async](
106108
region: String,
107109
table: String,
108110
prefix: String
109111
): F[Base64Hocon] =
110112
mkClient[F](region).use { dynamoDBClient =>
111-
// Assumes that the table is small and contains only the config
112-
val scanRequest = new ScanRequest().withTableName(table)
113+
val scanRequest = ScanRequest.builder().tableName(table).build()
113114
for {
114115
_ <- unsafeLogger.info(s"Retrieving enrichments in DynamoDB $region/$table/$prefix*")
115116
scanned <- withRetry(Async[F].blocking(dynamoDBClient.scan(scanRequest)))
116-
values = scanned.getItems().asScala.collect {
117-
case map if Option(map.get("id")).exists(_.getS.startsWith(prefix)) && map.containsKey("json") =>
118-
map.get("json").getS()
119-
}
117+
values = scanned
118+
.items()
119+
.asScala
120+
.collect {
121+
case map if Option(map.get("id")).exists(_.s.startsWith(prefix)) && map.containsKey("json") =>
122+
Option(map.get("json").s())
123+
}
124+
.flatten
120125
hocons <- values.toList
121126
.traverse { jsonStr =>
122127
Sync[F].delay(ConfigFactory.parseString(jsonStr))
@@ -133,13 +138,13 @@ object DynamoDbConfig {
133138
}
134139
}
135140

136-
private def mkClient[F[_]: Sync](region: String): Resource[F, AmazonDynamoDB] =
137-
Resource.make(Sync[F].delay {
138-
AmazonDynamoDBClientBuilder
139-
.standard()
140-
.withRegion(region)
141+
private def mkClient[F[_]: Sync](region: String): Resource[F, DynamoDbClient] =
142+
Resource.fromAutoCloseable(Sync[F].delay {
143+
DynamoDbClient
144+
.builder()
145+
.region(Region.of(region))
141146
.build()
142-
})(c => Sync[F].delay(c.shutdown))
147+
})
143148

144149
private def withRetry[F[_]: Async, A](f: F[A]): F[A] =
145150
retryingOnSomeErrors[A](retryPolicy[F], worthRetrying[F], onError[F])(f)
@@ -149,7 +154,7 @@ object DynamoDbConfig {
149154

150155
private def worthRetrying[F[_]: Applicative](e: Throwable): F[Boolean] =
151156
Applicative[F].pure(e match {
152-
case ace: AmazonClientException if ace.isRetryable => true
157+
case sdke: SdkException if sdke.retryable() => true
153158
case NonFatal(_) => false
154159
})
155160

‎modules/eventbridge/src/main/scala/com/snowplowanalytics/snowplow/enrich/eventbridge/Sink.scala‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@ object Sink {
5151
o.region.orElse(getRuntimeRegion) match {
5252
case Some(region) =>
5353
for {
54-
producer <- Resource.eval[F, EventBridgeClient](mkProducer(o, region))
54+
producer <- Resource.fromAutoCloseable(mkProducer(o, region))
5555
} yield (records: List[AttributedData[Array[Byte]]]) => writeToEventbridge(o, producer, toEventBridgeEvents(records, o))
5656
case None =>
5757
Resource.eval(Sync[F].raiseError(new RuntimeException(s"Region not found in the config and in the runtime")))
@@ -289,7 +289,6 @@ object Sink {
289289
size += entry.detailType().getBytes(StandardCharsets.UTF_8).length
290290
if (entry.detail() != null) size += entry.detail().getBytes(StandardCharsets.UTF_8).length
291291
if (entry.resources() != null) {
292-
import scala.collection.JavaConverters._
293292
for (resource <- entry.resources().asScala)
294293
if (resource != null) size += resource.getBytes(StandardCharsets.UTF_8).length
295294
}

‎modules/eventbridge/src/main/scala/com/snowplowanalytics/snowplow/enrich/eventbridge/package.scala‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,9 +12,9 @@
1212
*/
1313
package com.snowplowanalytics.snowplow.enrich
1414

15-
import com.amazonaws.regions.DefaultAwsRegionProviderChain
15+
import software.amazon.awssdk.regions.providers.DefaultAwsRegionProviderChain
1616

1717
package object eventbridge {
1818
def getRuntimeRegion: Option[String] =
19-
Option((new DefaultAwsRegionProviderChain).getRegion)
19+
Option(DefaultAwsRegionProviderChain.builder().build().getRegion).map(_.id())
2020
}

0 commit comments

Comments
 (0)