From fba430a4b03fb229fcdc05fef42e356ce7f3a36d Mon Sep 17 00:00:00 2001 From: abilan Date: Fri, 24 Mar 2023 17:11:19 -0400 Subject: [PATCH] GH-181, GH-182, GH-183: Migrate to AWS SDK v2 Fixes https://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis/issues/181 Fixes https://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis/issues/182 Fixes https://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis/issues/183 * Upgrade to the latest dependencies * Remove `SpringDynamoDBAdapterClient` since the DynamoDB Kinesis adapter is not recommended by AWS per se. * Fix docs to reflect new version logic * Remove `@Import({ContextCredentialsAutoConfiguration.class, ContextRegionProviderAutoConfiguration.class})` and fully rely on the auto-configuration --- README.adoc | 3 - pom.xml | 16 +- .../src/main/asciidoc/building.adoc | 6 +- .../src/main/asciidoc/contributing.adoc | 2 +- .../src/main/asciidoc/overview.adoc | 76 ++--- spring-cloud-stream-binder-kinesis/pom.xml | 20 +- .../kinesis/KinesisBinderHealthIndicator.java | 31 +- .../kinesis/KinesisMessageChannelBinder.java | 281 ++++++------------ .../adapter/SpringDynamoDBAdapterClient.java | 100 ------- .../config/KinesisBinderConfiguration.java | 125 ++++---- .../KinesisBinderConfigurationProperties.java | 2 +- .../properties/KinesisConsumerProperties.java | 10 - .../KinesisConsumerDestination.java | 4 +- .../KinesisProducerDestination.java | 4 +- .../KinesisStreamProvisioner.java | 234 ++++----------- .../kinesis/KinesisBinderFunctionalTests.java | 37 +-- .../binder/kinesis/KinesisBinderTests.java | 200 +++++-------- .../binder/kinesis/KinesisTestBinder.java | 80 ++--- .../kinesis/LocalstackContainerTest.java | 73 ++--- .../SpringDynamoDBAdapterClientTests.java | 200 ------------- .../KinesisBinderObservationTests.java | 8 +- .../KinesisStreamProvisionerTests.java | 281 +++++------------- 22 files changed, 525 insertions(+), 1268 deletions(-) delete mode 100644 spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/adapter/SpringDynamoDBAdapterClient.java delete mode 100644 spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/adapter/SpringDynamoDBAdapterClientTests.java diff --git a/README.adoc b/README.adoc index c8dae1c..2c7b57c 100644 --- a/README.adoc +++ b/README.adoc @@ -1,8 +1,5 @@ = Spring Cloud Stream Binder for AWS Kinesis -IMPORTANT: This project has been moved as a module to https://github.com/spring-cloud/spring-cloud-stream[Spring Cloud Stream] starting with its version `4.0.0`. -From now on it is recommended to create Kinesis binder-specific issues over there in Spring Cloud Stream. - This repository contains the code for the Spring Cloud Stream Binder implementation for AWS Kinesis. See https://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis/blob/main/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/overview.adoc[AWS Kinesis Binder Reference Manual] for more information. diff --git a/pom.xml b/pom.xml index d04fb1b..b7c0207 100644 --- a/pom.xml +++ b/pom.xml @@ -15,13 +15,12 @@ pom - 4.0.1 17 - 2.4.2 + 4.0.1 + 3.0.0-RC1 3.0.0-SNAPSHOT - 1.14.9 - 0.14.13 - 1.5.4 + 2.4.5 + 0.15.5 1.17.6 @@ -49,12 +48,7 @@ ${spring-integration-aws.version} - com.amazonaws - dynamodb-streams-kinesis-adapter - ${dynamodb-stream.version} - - - com.amazonaws + software.amazon.kinesis amazon-kinesis-client ${amazon-kinesis-client.version} diff --git a/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/building.adoc b/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/building.adoc index e8b37bb..ac7a4ad 100644 --- a/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/building.adoc +++ b/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/building.adoc @@ -7,8 +7,8 @@ To build the source you will need to install JDK {jdkversion}. -The build uses the Maven wrapper so you don't have to install a specific version of Maven. -To enable the tests, you should have https://github.com/mhart/kinesalite[Kinesalite] server running before building. +The build uses the Maven wrapper, so you don't have to install a specific version of Maven. +To enable the tests, you should have Docker running before building. The main build command is @@ -18,7 +18,7 @@ $ ./mvnw clean install You can also add `-DskipTests` if you like, to avoid running the tests. -NOTE: You can also install Maven (>=3.3.3) yourself and run the `mvn` command in place of `./mvnw` in the examples below. +NOTE: You can also install Maven (>=3.5.0) yourself and run the `mvn` command in place of `./mvnw` in the examples below. If you do that you also might need to add `-P spring` if your local Maven settings do not contain repository declarations for spring pre-release artifacts. NOTE: Be aware that you might need to increase the amount of memory available to Maven by setting a `MAVEN_OPTS` environment variable with a value like `-Xmx512m -XX:MaxPermSize=128m`. diff --git a/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/contributing.adoc b/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/contributing.adoc index 4a2b460..72039d8 100644 --- a/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/contributing.adoc +++ b/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/contributing.adoc @@ -1,7 +1,7 @@ [[contributing]] == Contributing -Spring Cloud is released under the non-restrictive Apache 2.0 license, and follows a very standard Github development process, using Github tracker for issues and merging pull requests into main. +Spring Cloud is released under the non-restrictive Apache 2.0 license, and follows a very standard GitHub development process, using GitHub tracker for issues and merging pull requests into main. If you want to contribute even something trivial please do not hesitate, but follow the guidelines below. === Sign the Contributor License Agreement diff --git a/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/overview.adoc index 0054b7e..4ed1ee7 100644 --- a/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/overview.adoc @@ -16,11 +16,13 @@ For using the AWS Kinesis Binder, you just need to add it to your Spring Cloud S ---- +NOTE: Starting with version `4.0`, this project is fully based on AWS SDK v2 and therefore has a lot of breaking changes and incompatibilities with the previous version. + == Kinesis Binder Overview The Spring Cloud Stream Binder for AWS Kinesis provides the binding implementation for the Spring Cloud Stream. This implementation uses Spring Integration AWS Kinesis Channel Adapters at its foundation. -The following captures how the Kinesis Binder implementation maps each of the configured destination to a AWS Kinesis Streams: +The following captures how the Kinesis Binder implementation maps each of the configured destination to AWS Kinesis Streams: .Kinesis Binder image::images/kinesis-binder.png[width=300,scaledwidth="50%"] @@ -45,8 +47,6 @@ The Spring Cloud Stream partition handling logic is excluded in case of AWS Kine On the consumer side the `instanceCount` and `instanceIndex` are used to distribute shards between consumers in group evenly. This has an effect only for regular `KinesisMessageDrivenChannelAdapter` which can assign specific shards for the target Kinesis consumer. -With Kinesis Client Library we can subscribe only to the provided stream and shards distribution is done by that client. -See more information in the Kinesis Client Library https://docs.aws.amazon.com/streams/latest/dev/developing-consumers-with-kcl.html[documentation]. == Consumer Groups Consumer groups are implemented with focus on High availability, Message ordering and guaranteed Message delivery in Spring cloud stream. @@ -115,7 +115,7 @@ It can be superseded by the `partitionCount` setting of the producer or by the v + Default: `1` kplKclEnabled:: - Enable the usage of https://docs.aws.amazon.com/streams/latest/dev/developing-consumers-with-kcl.html[Kinesis Client Library] / https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-kpl.html[Kinesis Producer Library] for all message consumption and production + Enable the usage of https://docs.aws.amazon.com/streams/latest/dev/kcl2-standard-consumer-java-example.html[Kinesis Client Library] / https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-kpl.html[Kinesis Producer Library] for all message consumption and production + Default: `false` + @@ -129,7 +129,7 @@ Default: `false` === MetadataStore Support for consumer groups is implemented using https://github.com/spring-projects/spring-integration-aws#metadata-store-for-amazon-dynamodb[DynamoDbMetadataStore]. -The `partitionKey` name used in the table is `KEY`. +The `partitionKey` name used in the table is `metadataKey`. This is not configurable. DynamoDB Checkpoint properties are prefixed with `spring.cloud.stream.kinesis.binder.checkpoint.` @@ -236,10 +236,6 @@ recordsLimit:: The maximum number of records to poll per `GetRecords` request. Must not be greater than `10000`. + -Default: `10000` -idleBetweenPolls:: - The sleep interval used in the main loop between shards polling cycles, in milliseconds. Must not be less than `250`. -+ Default: `1000` consumerBackoff:: The amount of time the consumer will wait to attempt another `GetRecords` operation after a read with no results, in milliseconds. @@ -257,16 +253,6 @@ It is by default 1 day, and it can be extended to 7 days. This happens only for new consumer groups. Any subsequent starts of the consumer in the same group are adjusted according the stored checkpoint via `AFTER_SEQUENCE_NUMBER` iterator type. -dynamoDbStreams:: - The `boolean` flag indicating that Kinesis consumer channel adapter should adapt DynamoDB Streams functionality instead of regular Kinesis streams. -The `spring.cloud.stream.bindings..destination` value must be a DynamoDB table name. -Default: `false`. - -Starting with version 2.0.1, beans of `KinesisClientLibConfiguration` type can be provided in the application context to have a full control over Kinesis Client Library configuration options. -The stream and consumer group (plus workerId) must be provided in the respective `KinesisClientLibConfiguration` bean. -When `KclMessageDrivenChannelAdapter` endpoint is configured in the binder, it selects an appropriate `KinesisClientLibConfiguration` from the application context according a destination (stream) for binding. -If there is no `KinesisClientLibConfiguration` bean for its stream, the `KclMessageDrivenChannelAdapter` falls back to original configuration with defaults options for its internal `KinesisClientLibConfiguration`. - shardId:: An explicit shard id to consume from. @@ -290,26 +276,20 @@ sendTimeout:: Default: `10000` Also, if you'd like to produce a batch of records into Kinesis stream, the message payload must be as a `PutRecordsRequest` instance and general Spring Cloud Stream producer property `useNativeEncoding` must be set to `true`, so Spring Cloud Stream won't try to convert a `PutRecordsRequest` into a `byte[]`. The content of the `PutRecordsRequest` is now end-user responsibility. + [[kinesis-error-channels]] == Error Channels The binder can be configured to send producer exceptions to an error channel. See https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream.html#spring-cloud-stream-overview-error-handling[the section on Spring Cloud Stream error handling] for more information. -The payload of the `ErrorMessage` for a send failure is an `AwsRequestFailureException` with properties: +The payload of the `ErrorMessage` for a `send` failure is an `AwsRequestFailureException` with properties: * `failedMessage` - the spring-messaging `Message` that failed to be sent. -* `request` - the raw `AmazonWebServiceRequest` (either `PutRecordRequest` or `PutRecordsRequest`) that was created from the `failedMessage`. +* `request` - the raw `AwsRequest` (either `PutRecordRequest` or `PutRecordsRequest`) that was created from the `failedMessage`. There is no automatic handling of these exceptions (such as sending to a dead letter queue), but you can consume these exceptions with your own Spring Integration flow. -[[dynamodb-streams]] -== DynamoDB Streams - -Starting with version 1.2, the `KinesisMessageChannelBinder` supports DynamoDB Streams for Kinesis consumer channel adapter. -The functionality is based on the https://github.com/awslabs/dynamodb-streams-kinesis-adapter[DynamoDB Streams Kinesis Adapter] and every Kinesis consumer endpoint selects an appropriate Kinesis Client according the `spring.cloud.stream.kinesis.bindings..consumer.dynamoDbStreams` boolean flag. -In this case a value for `spring.cloud.stream.bindings..destination` must be a DynamoDB table name. - [[optional-resources]] == Optional Resources @@ -401,8 +381,7 @@ If you're going to allow spring-cloud-stream-binder-kinesis to create the resour == Running locally with localstack Sometimes we don't have the necessary permissions to connect to the real Kinesis and DynamoDB from our developer's machine. -In moments like this, it's pretty useful to setup Localstack in your project, so you can run everything locally, without -having to worry about permissions and enterprise restrictions. +In moments like this, it's pretty useful to setup Localstack in your project, so you can run everything locally, without having to worry about permissions and enterprise restrictions. Create a *docker-compose.yaml* file, in the root of your project, to quickly start localstack @@ -412,7 +391,7 @@ version: '3.5' services: localstack: - image: localstack/localstack:0.12.10 + image: localstack/localstack:1.4.0 environment: - AWS_DEFAULT_REGION=sa-east-1 - EDGE_PORT=4566 @@ -471,8 +450,8 @@ aws --endpoint-url=http://localhost:4566 kinesis put-record --stream-name my-tes [[telling-the-binder-to-use-your-local-endpoint]] === Telling the binder to use your local endpoint -By default, the Kinesis and DynamoDB Client will try to hit the real AWS Endpoint. To change this behavior, -you have to declare a new @Bean, and override the endpoint. +By default, the Kinesis and DynamoDB Client will try to hit the real AWS Endpoint. +To change this behavior, you have to declare a new `@Bean`, and override the endpoint. For example: @@ -481,16 +460,18 @@ For example: @Configuration @Profile("local") class DynamoDBConfigLocal { + @Value("\${cloud.aws.region.static}") - val region: String = "" + lateinit val region: String private val endpointUrl: String = "http://localhost:4566" @Bean @Primary - fun amazonDynamoDBAsync(): AmazonDynamoDBAsync { - return AmazonDynamoDBAsyncClientBuilder.standard() - .withEndpointConfiguration(AwsClientBuilder.EndpointConfiguration(this.endpointUrl, region)) + fun amazonDynamoDBAsync(): DynamoDbAsync { + return DynamoDbAsyncClient.builder() + .region(region) + .endpointOverride(Url(endpointUrl)) .build() } @@ -502,17 +483,18 @@ class DynamoDBConfigLocal { @Configuration @Profile("local") class KinesisConfigLocal { + @Value("\${cloud.aws.region.static}") - val region: String = "" + lateinit val region: String private val endpointUrl: String = "http://localhost:4566" @Bean - fun amazonKinesis(awsCredentialsProvider: AWSCredentialsProvider): AmazonKinesisAsync { - return AmazonKinesisAsyncClientBuilder - .standard() - .withCredentials(awsCredentialsProvider) - .withEndpointConfiguration(AwsClientBuilder.EndpointConfiguration(endpointUrl, region)) + fun amazonKinesis(awsCredentialsProvider: AwsCredentialsProvider): KinesisAsyncClient { + return KinesisAsyncClient.builder() + .credentialsProvider(awsCredentialsProvider) + .region(region) + .endpointOverride(Url(endpointUrl)) .build() } @@ -522,10 +504,12 @@ class KinesisConfigLocal { Now, remember to pass the following environment variables when running locally: [source, shell script] ---- -SPRING_PROFILES_ACTIVE=local AWS_CBOR_DISABLE=true gradle bootRun +SPRING_PROFILES_ACTIVE=local gradle bootRun ---- -This will make sure that these beans are only instantiated when running locally, and will also disable CBOR, which is not -supported for the localstack's kinesis stream. +This will make sure that these beans are only instantiated when running locally, which is not supported for the localstack's kinesis stream. + +Another way to test against LocalStack is to use `Testcontainers` with a `localstack/localstack` image container. +The `LocalstackContainerTest` interface in this project can be a good sample how to configure container and how to use AWS clients. [[health-indicator]] == Kinesis Binder Health Indicator diff --git a/spring-cloud-stream-binder-kinesis/pom.xml b/spring-cloud-stream-binder-kinesis/pom.xml index 85d8c9f..55e2359 100644 --- a/spring-cloud-stream-binder-kinesis/pom.xml +++ b/spring-cloud-stream-binder-kinesis/pom.xml @@ -17,7 +17,7 @@ - com.amazonaws + software.amazon.kinesis amazon-kinesis-client @@ -34,21 +34,7 @@ io.awspring.cloud - spring-cloud-starter-aws - - - com.amazonaws - aws-java-sdk-dynamodb - - - - com.amazonaws - aws-java-sdk-kinesis - - - - com.amazonaws - dynamodb-streams-kinesis-adapter + spring-cloud-aws-starter @@ -106,7 +92,7 @@ io.micrometer micrometer-tracing-bom - 1.1.0-M1 + 1.1.0-M2 pom import diff --git a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderHealthIndicator.java b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderHealthIndicator.java index c509107..56352e3 100644 --- a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderHealthIndicator.java +++ b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderHealthIndicator.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2021 the original author or authors. + * Copyright 2017-2023 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,11 +18,11 @@ package org.springframework.cloud.stream.binder.kinesis; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CompletionException; import java.util.concurrent.TimeUnit; -import com.amazonaws.services.kinesis.AmazonKinesisAsync; -import com.amazonaws.services.kinesis.model.LimitExceededException; -import com.amazonaws.services.kinesis.model.ListShardsRequest; +import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; +import software.amazon.awssdk.services.kinesis.model.LimitExceededException; import org.springframework.boot.actuate.health.Health; import org.springframework.boot.actuate.health.HealthIndicator; @@ -42,26 +42,29 @@ public class KinesisBinderHealthIndicator implements HealthIndicator { @Override public Health health() { - AmazonKinesisAsync amazonKinesis = this.kinesisMessageChannelBinder.getAmazonKinesis(); + KinesisAsyncClient amazonKinesis = this.kinesisMessageChannelBinder.getAmazonKinesis(); List streamsInUse = new ArrayList<>(this.kinesisMessageChannelBinder.getStreamsInUse()); for (String stream : streamsInUse) { while (true) { try { - amazonKinesis.listShards(new ListShardsRequest().withStreamName(stream).withMaxResults(1)); + amazonKinesis.listShards(request -> request.streamName(stream).maxResults(1)).join(); break; } - catch (LimitExceededException ex) { - try { - TimeUnit.SECONDS.sleep(1); + catch (CompletionException ex) { + Throwable cause = ex.getCause(); + if (cause instanceof LimitExceededException) { + try { + TimeUnit.SECONDS.sleep(1); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return Health.down(ex).build(); + } } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); + else { return Health.down(ex).build(); } } - catch (Exception ex) { - return Health.down(ex).build(); - } } } return Health.up().build(); diff --git a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java index 3ba706d..a161132 100644 --- a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java @@ -16,34 +16,27 @@ package org.springframework.cloud.stream.binder.kinesis; -import java.time.Duration; +import java.time.Instant; import java.util.ArrayList; import java.util.Collections; import java.util.Date; import java.util.HashSet; import java.util.List; -import java.util.Objects; import java.util.Set; import java.util.UUID; -import com.amazonaws.ClientConfiguration; -import com.amazonaws.auth.AWSCredentialsProvider; -import com.amazonaws.services.cloudwatch.AmazonCloudWatch; -import com.amazonaws.services.dynamodbv2.AmazonDynamoDB; -import com.amazonaws.services.dynamodbv2.AmazonDynamoDBStreams; -import com.amazonaws.services.dynamodbv2.model.DescribeTableResult; -import com.amazonaws.services.dynamodbv2.streamsadapter.AmazonDynamoDBStreamsAdapterClient; -import com.amazonaws.services.kinesis.AmazonKinesis; -import com.amazonaws.services.kinesis.AmazonKinesisAsync; -import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream; -import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration; -import com.amazonaws.services.kinesis.clientlibrary.lib.worker.SimpleRecordsFetcherFactory; -import com.amazonaws.services.kinesis.model.InvalidArgumentException; -import com.amazonaws.services.kinesis.model.Shard; -import com.amazonaws.services.kinesis.model.ShardIteratorType; import com.amazonaws.services.kinesis.producer.KinesisProducer; import com.amazonaws.services.kinesis.producer.KinesisProducerConfiguration; import io.micrometer.observation.ObservationRegistry; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; +import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient; +import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; +import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; +import software.amazon.awssdk.services.kinesis.model.InvalidArgumentException; +import software.amazon.awssdk.services.kinesis.model.Shard; +import software.amazon.awssdk.services.kinesis.model.ShardIteratorType; +import software.amazon.kinesis.common.InitialPositionInStream; +import software.amazon.kinesis.common.InitialPositionInStreamExtended; import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; import org.springframework.cloud.stream.binder.BinderHeaders; @@ -52,7 +45,6 @@ import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy; -import org.springframework.cloud.stream.binder.kinesis.adapter.SpringDynamoDBAdapterClient; import org.springframework.cloud.stream.binder.kinesis.properties.KinesisBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kinesis.properties.KinesisConsumerProperties; import org.springframework.cloud.stream.binder.kinesis.properties.KinesisExtendedBindingProperties; @@ -61,7 +53,6 @@ import org.springframework.cloud.stream.binder.kinesis.provisioning.KinesisConsu import org.springframework.cloud.stream.binder.kinesis.provisioning.KinesisStreamProvisioner; import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.cloud.stream.provisioning.ProducerDestination; -import org.springframework.cloud.stream.provisioning.ProvisioningException; import org.springframework.expression.EvaluationContext; import org.springframework.integration.aws.inbound.kinesis.KclMessageDrivenChannelAdapter; import org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter; @@ -115,15 +106,13 @@ public class KinesisMessageChannelBinder extends private final KinesisBinderConfigurationProperties configurationProperties; - private final AmazonDynamoDBStreamsAdapterClient dynamoDBStreamsAdapter; + private final KinesisAsyncClient amazonKinesis; - private final AmazonKinesisAsync amazonKinesis; + private final AwsCredentialsProvider awsCredentialsProvider; - private final AWSCredentialsProvider awsCredentialsProvider; + private final CloudWatchAsyncClient cloudWatchClient; - private final AmazonCloudWatch cloudWatchClient; - - private final AmazonDynamoDB dynamoDBClient; + private final DynamoDbAsyncClient dynamoDBClient; private final String[] headersToEmbed; @@ -140,16 +129,13 @@ public class KinesisMessageChannelBinder extends private EvaluationContext evaluationContext; - private List kinesisClientLibConfigurations; - private ObservationRegistry observationRegistry = ObservationRegistry.NOOP; public KinesisMessageChannelBinder(KinesisBinderConfigurationProperties configurationProperties, - KinesisStreamProvisioner provisioningProvider, AmazonKinesisAsync amazonKinesis, - AWSCredentialsProvider awsCredentialsProvider, - @Nullable AmazonDynamoDB dynamoDBClient, - @Nullable AmazonDynamoDBStreams dynamoDBStreams, - @Nullable AmazonCloudWatch cloudWatchClient) { + KinesisStreamProvisioner provisioningProvider, KinesisAsyncClient amazonKinesis, + AwsCredentialsProvider awsCredentialsProvider, + @Nullable DynamoDbAsyncClient dynamoDBClient, + @Nullable CloudWatchAsyncClient cloudWatchClient) { super(new String[0], provisioningProvider); Assert.notNull(amazonKinesis, "'amazonKinesis' must not be null"); @@ -160,12 +146,6 @@ public class KinesisMessageChannelBinder extends this.dynamoDBClient = dynamoDBClient; this.awsCredentialsProvider = awsCredentialsProvider; - if (dynamoDBStreams != null) { - this.dynamoDBStreamsAdapter = new SpringDynamoDBAdapterClient(dynamoDBStreams); - } - else { - this.dynamoDBStreamsAdapter = null; - } this.headersToEmbed = headersToMap(configurationProperties); } @@ -185,10 +165,6 @@ public class KinesisMessageChannelBinder extends this.kinesisProducerConfiguration = kinesisProducerConfiguration; } - public void setKinesisClientLibConfigurations(List kinesisClientLibConfigurations) { - this.kinesisClientLibConfigurations = kinesisClientLibConfigurations; - } - public void setObservationRegistry(@Nullable ObservationRegistry observationRegistry) { this.observationRegistry = observationRegistry; } @@ -213,7 +189,7 @@ public class KinesisMessageChannelBinder extends return this.extendedBindingProperties.getExtendedPropertiesEntryClass(); } - public AmazonKinesisAsync getAmazonKinesis() { + public KinesisAsyncClient getAmazonKinesis() { return this.amazonKinesis; } @@ -234,8 +210,8 @@ public class KinesisMessageChannelBinder extends @Override protected MessageHandler createProducerMessageHandler(ProducerDestination destination, - ExtendedProducerProperties producerProperties, - MessageChannel errorChannel) { + ExtendedProducerProperties producerProperties, MessageChannel channel, + @Nullable MessageChannel errorChannel) { FunctionExpression> partitionKeyExpression = new FunctionExpression<>((m) -> @@ -251,16 +227,33 @@ public class KinesisMessageChannelBinder extends messageHandler = createKinesisMessageHandler(destination, partitionKeyExpression, producerProperties.getExtension().isEmbedHeaders()); } - messageHandler.setSync(producerProperties.getExtension().isSync()); + messageHandler.setAsync(!producerProperties.getExtension().isSync()); messageHandler.setSendTimeout(producerProperties.getExtension().getSendTimeout()); - messageHandler.setFailureChannel(errorChannel); messageHandler.setBeanFactory(getBeanFactory()); + if (errorChannel != null) { + ((InterceptableChannel) channel) + .addInterceptor(new ChannelInterceptor() { + + @Override + public Message preSend(Message message, MessageChannel channel) { + return MessageBuilder.fromMessage(message).setErrorChannel(errorChannel).build(); + } + + }); + } + this.streamsInUse.add(destination.getName()); return messageHandler; } + @Override + protected MessageHandler createProducerMessageHandler(ProducerDestination destination, + ExtendedProducerProperties producerProperties, MessageChannel errorChannel) { + return null; + } + private AbstractAwsMessageHandler createKinesisMessageHandler(ProducerDestination destination, FunctionExpression> partitionKeyExpression, boolean embedHeaders) { @@ -276,8 +269,7 @@ public class KinesisMessageChannelBinder extends private AbstractAwsMessageHandler createKplMessageHandler(ProducerDestination destination, FunctionExpression> partitionKeyExpression, boolean embedHeaders) { - final KplMessageHandler messageHandler; - messageHandler = new KplMessageHandler(new KinesisProducer(this.kinesisProducerConfiguration)); + KplMessageHandler messageHandler = new KplMessageHandler(new KinesisProducer(this.kinesisProducerConfiguration)); messageHandler.setStream(destination.getName()); messageHandler.setPartitionKeyExpression(partitionKeyExpression); if (embedHeaders) { @@ -332,34 +324,18 @@ public class KinesisMessageChannelBinder extends String group, ExtendedConsumerProperties properties) { - ConsumerDestination destinationToUse = destination; - - if (properties.getExtension().isDynamoDbStreams()) { - DescribeTableResult describeTableResult = this.dynamoDBClient.describeTable(destinationToUse.getName()); - String latestStreamArn = describeTableResult.getTable().getLatestStreamArn(); - if (StringUtils.hasText(latestStreamArn)) { - destinationToUse = new KinesisConsumerDestination(latestStreamArn, Collections.emptyList()); - } - else { - throw new ProvisioningException("The DynamoDB table [" - + destinationToUse.getName() - + "] doesn't have Streams enabled."); - } - } - else { - this.streamsInUse.add(destinationToUse.getName()); - } + this.streamsInUse.add(destination.getName()); MessageProducerSupport adapter; if (this.configurationProperties.isKplKclEnabled()) { - adapter = createKclConsumerEndpoint(destinationToUse, group, properties); + adapter = createKclConsumerEndpoint(destination, group, properties); } else { - adapter = createKinesisConsumerEndpoint(destinationToUse, group, properties); + adapter = createKinesisConsumerEndpoint(destination, group, properties); } adapter.registerObservationRegistry(this.observationRegistry); - adapter.setComponentName(String.format("Consumer for [%s]", destinationToUse.getName())); + adapter.setComponentName(String.format("Consumer for [%s]", destination.getName())); return adapter; } @@ -374,23 +350,55 @@ public class KinesisMessageChannelBinder extends "Ignoring 'shardId' property"); } - AmazonKinesis amazonKinesisClient = - kinesisConsumerProperties.isDynamoDbStreams() - ? this.dynamoDBStreamsAdapter - : this.amazonKinesis; - - String stream = destination.getName(); - - KinesisClientLibConfiguration kinesisClientLibConfiguration = - obtainKinesisClientLibConfiguration(properties.getExtension(), stream, group); - KclMessageDrivenChannelAdapter adapter = - new KclMessageDrivenChannelAdapter(kinesisClientLibConfiguration, amazonKinesisClient, - this.cloudWatchClient, this.dynamoDBClient); - String consumerGroup = kinesisClientLibConfiguration.getApplicationName(); + new KclMessageDrivenChannelAdapter(this.amazonKinesis, this.cloudWatchClient, this.dynamoDBClient, + destination.getName()); + boolean anonymous = !StringUtils.hasText(group); + String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID() : group; + + String workerId = + kinesisConsumerProperties.getWorkerId() != null + ? kinesisConsumerProperties.getWorkerId() + : UUID.randomUUID().toString(); + + String shardIteratorType = kinesisConsumerProperties.getShardIteratorType(); + + InitialPositionInStreamExtended kinesisShardOffset = + InitialPositionInStreamExtended.newInitialPosition(InitialPositionInStream.LATEST); + + if (StringUtils.hasText(shardIteratorType)) { + String[] typeValue = shardIteratorType.split(":", 2); + ShardIteratorType iteratorType = ShardIteratorType.valueOf(typeValue[0]); + if (typeValue.length > 1) { + if (ShardIteratorType.AT_TIMESTAMP.equals(iteratorType)) { + kinesisShardOffset = + InitialPositionInStreamExtended.newInitialPositionAtTimestamp( + new Date(Long.parseLong(typeValue[1]))); + } + else { + throw new IllegalArgumentException("The KCL does not support 'AT_SEQUENCE_NUMBER' " + + "or 'AFTER_SEQUENCE_NUMBER' initial position in stream."); + } + } + else { + kinesisShardOffset = + InitialPositionInStreamExtended.newInitialPosition( + InitialPositionInStream.valueOf(iteratorType.name())); + } + } + + kinesisShardOffset = + anonymous || StringUtils.hasText(shardIteratorType) + ? kinesisShardOffset + : InitialPositionInStreamExtended.newInitialPosition(InitialPositionInStream.TRIM_HORIZON); + + adapter.setConsumerGroup(consumerGroup); + adapter.setWorkerId(workerId); + adapter.setStreamInitialSequence(kinesisShardOffset); adapter.setCheckpointMode(kinesisConsumerProperties.getCheckpointMode()); adapter.setCheckpointsInterval(kinesisConsumerProperties.getCheckpointInterval()); + adapter.setConsumerBackoff(kinesisConsumerProperties.getConsumerBackoff()); adapter.setListenerMode(kinesisConsumerProperties.getListenerMode()); if (properties.getExtension().isEmbedHeaders()) { adapter.setEmbeddedHeadersMapper(new EmbeddedJsonHeadersMessageMapper()); @@ -411,103 +419,15 @@ public class KinesisMessageChannelBinder extends return adapter; } - private KinesisClientLibConfiguration obtainKinesisClientLibConfiguration( - KinesisConsumerProperties properties, String stream, String group) { - - KinesisClientLibConfiguration candidate = null; - for (KinesisClientLibConfiguration conf : this.kinesisClientLibConfigurations) { - if (stream.equals(conf.getStreamName())) { - candidate = conf; - if (Objects.equals(group, conf.getApplicationName())) { - break; - } - } - } - - if (candidate == null) { - boolean anonymous = !StringUtils.hasText(group); - String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID() : group; - - candidate = new KinesisClientLibConfiguration(consumerGroup, - stream, - null, - null, - InitialPositionInStream.LATEST, - this.awsCredentialsProvider, - null, - null, - KinesisClientLibConfiguration.DEFAULT_FAILOVER_TIME_MILLIS, - properties.getWorkerId() != null ? properties.getWorkerId() : UUID.randomUUID().toString(), - KinesisClientLibConfiguration.DEFAULT_MAX_RECORDS, - properties.getIdleBetweenPolls(), - false, - KinesisClientLibConfiguration.DEFAULT_PARENT_SHARD_POLL_INTERVAL_MILLIS, - KinesisClientLibConfiguration.DEFAULT_SHARD_SYNC_INTERVAL_MILLIS, - KinesisClientLibConfiguration.DEFAULT_CLEANUP_LEASES_UPON_SHARDS_COMPLETION, - new ClientConfiguration(), - new ClientConfiguration(), - new ClientConfiguration(), - properties.getConsumerBackoff(), - KinesisClientLibConfiguration.DEFAULT_METRICS_BUFFER_TIME_MILLIS, - KinesisClientLibConfiguration.DEFAULT_METRICS_MAX_QUEUE_SIZE, - KinesisClientLibConfiguration.DEFAULT_VALIDATE_SEQUENCE_NUMBER_BEFORE_CHECKPOINTING, - null, - KinesisClientLibConfiguration.DEFAULT_SHUTDOWN_GRACE_MILLIS, - KinesisClientLibConfiguration.DEFAULT_DDB_BILLING_MODE, - new SimpleRecordsFetcherFactory(), - Duration.ofMinutes(1).toMillis(), - Duration.ofMinutes(5).toMillis(), - Duration.ofMinutes(30).toMillis()); - - String shardIteratorType = properties.getShardIteratorType(); - - KinesisShardOffset kinesisShardOffset = KinesisShardOffset.latest(); - - if (StringUtils.hasText(shardIteratorType)) { - String[] typeValue = shardIteratorType.split(":", 2); - ShardIteratorType iteratorType = ShardIteratorType.valueOf(typeValue[0]); - kinesisShardOffset = new KinesisShardOffset(iteratorType); - if (typeValue.length > 1) { - if (ShardIteratorType.AT_TIMESTAMP.equals(iteratorType)) { - kinesisShardOffset - .setTimestamp(new Date(Long.parseLong(typeValue[1]))); - } - else { - kinesisShardOffset.setSequenceNumber(typeValue[1]); - } - } - } - - kinesisShardOffset = - anonymous || StringUtils.hasText(shardIteratorType) - ? kinesisShardOffset - : KinesisShardOffset.trimHorizon(); - - if (kinesisShardOffset.getIteratorType().equals(ShardIteratorType.AT_TIMESTAMP)) { - candidate.withTimestampAtInitialPositionInStream(kinesisShardOffset.getTimestamp()); - } - else if (kinesisShardOffset.getIteratorType().equals(ShardIteratorType.AT_SEQUENCE_NUMBER) || - kinesisShardOffset.getIteratorType().equals(ShardIteratorType.AFTER_SEQUENCE_NUMBER)) { - - throw new IllegalArgumentException("The KCL does not support 'AT_SEQUENCE_NUMBER' " + - "or 'AFTER_SEQUENCE_NUMBER' initial position in stream."); - } - else { - candidate.withInitialPositionInStream( - InitialPositionInStream.valueOf(kinesisShardOffset.getIteratorType().name())); - } - } - - return candidate; - } - private MessageProducerSupport createKinesisConsumerEndpoint(ConsumerDestination destination, String group, ExtendedConsumerProperties properties) { KinesisConsumerProperties kinesisConsumerProperties = properties.getExtension(); if (properties.getInstanceCount() > 1 && properties.getExtension().getShardId() != null) { - throw new InvalidArgumentException("'instanceCount' more than 1 and 'shardId' cannot be provided together."); + throw InvalidArgumentException.builder() + .message("'instanceCount' more than 1 and 'shardId' cannot be provided together.") + .build(); } Set shardOffsets = null; @@ -523,7 +443,7 @@ public class KinesisMessageChannelBinder extends if (typeValue.length > 1) { if (ShardIteratorType.AT_TIMESTAMP.equals(iteratorType)) { kinesisShardOffset - .setTimestamp(new Date(Long.parseLong(typeValue[1]))); + .setTimestamp(Instant.ofEpochMilli(Long.parseLong(typeValue[1]))); } else { kinesisShardOffset.setSequenceNumber(typeValue[1]); @@ -540,7 +460,7 @@ public class KinesisMessageChannelBinder extends if ((i % properties.getInstanceCount()) == properties.getInstanceIndex()) { KinesisShardOffset shardOffset = new KinesisShardOffset(kinesisShardOffset); shardOffset.setStream(destination.getName()); - shardOffset.setShard(shards.get(i).getShardId()); + shardOffset.setShard(shards.get(i).shardId()); shardOffsets.add(shardOffset); } } @@ -548,24 +468,19 @@ public class KinesisMessageChannelBinder extends KinesisMessageDrivenChannelAdapter adapter; - AmazonKinesis amazonKinesisClient = - kinesisConsumerProperties.isDynamoDbStreams() - ? this.dynamoDBStreamsAdapter - : this.amazonKinesis; - String shardId = kinesisConsumerProperties.getShardId(); if (CollectionUtils.isEmpty(shardOffsets) && shardId == null) { - adapter = new KinesisMessageDrivenChannelAdapter(amazonKinesisClient, destination.getName()); + adapter = new KinesisMessageDrivenChannelAdapter(this.amazonKinesis, destination.getName()); } else if (shardId != null) { KinesisShardOffset shardOffset = new KinesisShardOffset(kinesisShardOffset); shardOffset.setStream(destination.getName()); shardOffset.setShard(shardId); - adapter = new KinesisMessageDrivenChannelAdapter(amazonKinesisClient, shardOffset); + adapter = new KinesisMessageDrivenChannelAdapter(this.amazonKinesis, shardOffset); } else { - adapter = new KinesisMessageDrivenChannelAdapter(amazonKinesisClient, + adapter = new KinesisMessageDrivenChannelAdapter(this.amazonKinesis, shardOffsets.toArray(new KinesisShardOffset[0])); } diff --git a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/adapter/SpringDynamoDBAdapterClient.java b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/adapter/SpringDynamoDBAdapterClient.java deleted file mode 100644 index 4b529d4..0000000 --- a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/adapter/SpringDynamoDBAdapterClient.java +++ /dev/null @@ -1,100 +0,0 @@ -/* - * Copyright 2020-2020 the original author or authors. - * - * Licensed 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 - * - * https://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.springframework.cloud.stream.binder.kinesis.adapter; - -import com.amazonaws.services.dynamodbv2.AmazonDynamoDBStreams; -import com.amazonaws.services.dynamodbv2.model.AmazonDynamoDBException; -import com.amazonaws.services.dynamodbv2.streamsadapter.AmazonDynamoDBStreamsAdapterClient; -import com.amazonaws.services.kinesis.model.DescribeStreamRequest; -import com.amazonaws.services.kinesis.model.InvalidArgumentException; -import com.amazonaws.services.kinesis.model.ListShardsRequest; -import com.amazonaws.services.kinesis.model.ListShardsResult; -import com.amazonaws.services.kinesis.model.ResourceNotFoundException; -import com.amazonaws.services.kinesis.model.StreamDescription; - -/** - * This is Spring Cloud DynamoDB Adapter to be able to support {@code ListShards} operations. - * - * @author Asiel Caballero - * - * @since 2.0.3 - * - * @see ListShards - */ -public class SpringDynamoDBAdapterClient extends AmazonDynamoDBStreamsAdapterClient { - - private static final String SEPARATOR = "!!##%%"; - - public SpringDynamoDBAdapterClient(AmazonDynamoDBStreams amazonDynamoDBStreams) { - super(amazonDynamoDBStreams); - } - - /** - * List shards for a DynamoDB Stream using its {@code DescribeStream} API, as they don't support - * {@code ListShards} operations. Returns the result adapted to use the AmazonKinesis model. - * @param request Container for the necessary parameters to execute the ListShards service method - * @return The response from the DescribeStream service method, adapted for use with the AmazonKinesis model - */ - @Override - public ListShardsResult listShards(ListShardsRequest request) { - try { - if (request.getNextToken() != null && request.getStreamName() != null) { - throw new InvalidArgumentException("NextToken and StreamName cannot be provided together."); - } - - String streamName = request.getStreamName(); - String exclusiveStartShardId = request.getExclusiveStartShardId(); - - if (request.getNextToken() != null) { - String[] split = request.getNextToken().split(SEPARATOR); - - if (split.length != 2) { - throw new InvalidArgumentException("Invalid ShardIterator"); - } - - streamName = split[0]; - exclusiveStartShardId = split[1]; - } - - DescribeStreamRequest dsr = new DescribeStreamRequest() - .withStreamName(streamName) - .withExclusiveStartShardId(exclusiveStartShardId) - .withLimit(request.getMaxResults()); - StreamDescription streamDescription = describeStream(dsr).getStreamDescription(); - - ListShardsResult result = new ListShardsResult() - .withShards(streamDescription.getShards()); - - if (streamDescription.getHasMoreShards()) { - result.withNextToken(buildFakeNextToken(streamName, - streamDescription.getShards().get(streamDescription.getShards().size() - 1).getShardId())); - } - - return result; - } - catch (AmazonDynamoDBException ex) { - ResourceNotFoundException resourceEx = new ResourceNotFoundException(ex.getMessage()); - resourceEx.setStackTrace(ex.getStackTrace()); - throw resourceEx; - } - } - - private String buildFakeNextToken(String streamName, String lastShard) { - return lastShard != null ? streamName + SEPARATOR + lastShard : null; - } - -} diff --git a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/config/KinesisBinderConfiguration.java b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/config/KinesisBinderConfiguration.java index 45b6165..c708f9b 100644 --- a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/config/KinesisBinderConfiguration.java +++ b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/config/KinesisBinderConfiguration.java @@ -19,27 +19,23 @@ package org.springframework.cloud.stream.binder.kinesis.config; import java.util.List; import java.util.Set; +import com.amazonaws.auth.AWSCredentials; import com.amazonaws.auth.AWSCredentialsProvider; -import com.amazonaws.services.cloudwatch.AmazonCloudWatch; -import com.amazonaws.services.cloudwatch.AmazonCloudWatchAsync; -import com.amazonaws.services.cloudwatch.AmazonCloudWatchAsyncClientBuilder; -import com.amazonaws.services.dynamodbv2.AmazonDynamoDB; -import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync; -import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsyncClientBuilder; -import com.amazonaws.services.dynamodbv2.AmazonDynamoDBStreams; -import com.amazonaws.services.dynamodbv2.AmazonDynamoDBStreamsClientBuilder; -import com.amazonaws.services.kinesis.AmazonKinesisAsync; -import com.amazonaws.services.kinesis.AmazonKinesisAsyncClientBuilder; -import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration; +import com.amazonaws.auth.BasicAWSCredentials; import com.amazonaws.services.kinesis.producer.KinesisProducerConfiguration; -import io.awspring.cloud.autoconfigure.context.ContextCredentialsAutoConfiguration; -import io.awspring.cloud.autoconfigure.context.ContextRegionProviderAutoConfiguration; -import io.awspring.cloud.core.region.RegionProvider; import io.micrometer.observation.ObservationRegistry; +import software.amazon.awssdk.auth.credentials.AwsCredentials; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.regions.providers.AwsRegionProvider; +import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient; +import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; +import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.actuate.autoconfigure.health.ConditionalOnEnabledHealthIndicator; import org.springframework.boot.actuate.health.HealthIndicator; +import org.springframework.boot.autoconfigure.AutoConfiguration; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; @@ -56,7 +52,6 @@ import org.springframework.cloud.stream.config.ConsumerEndpointCustomizer; import org.springframework.cloud.stream.config.ProducerMessageHandlerCustomizer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.context.annotation.Import; import org.springframework.integration.aws.lock.DynamoDbLockRegistry; import org.springframework.integration.aws.lock.DynamoDbLockRepository; import org.springframework.integration.aws.metadata.DynamoDbMetadataStore; @@ -73,28 +68,27 @@ import org.springframework.integration.support.locks.LockRegistry; * @author Arnaud Lecollaire * @author Asiel Caballero */ -@Configuration(proxyBeanMethods = false) +@AutoConfiguration @ConditionalOnMissingBean(Binder.class) @EnableConfigurationProperties({KinesisBinderConfigurationProperties.class, KinesisExtendedBindingProperties.class}) -@Import({ContextCredentialsAutoConfiguration.class, ContextRegionProviderAutoConfiguration.class}) public class KinesisBinderConfiguration { private final KinesisBinderConfigurationProperties configurationProperties; - private final AWSCredentialsProvider awsCredentialsProvider; + private final AwsCredentialsProvider awsCredentialsProvider; - private final String region; + private final Region region; private final boolean hasInputs; public KinesisBinderConfiguration(KinesisBinderConfigurationProperties configurationProperties, - AWSCredentialsProvider awsCredentialsProvider, - RegionProvider regionProvider, + AwsCredentialsProvider awsCredentialsProvider, + AwsRegionProvider regionProvider, List bindables) { this.configurationProperties = configurationProperties; this.awsCredentialsProvider = awsCredentialsProvider; - this.region = regionProvider.getRegion().getName(); + this.region = regionProvider.getRegion(); this.hasInputs = bindables.stream() .map(Bindable::getInputs) @@ -105,25 +99,25 @@ public class KinesisBinderConfiguration { @Bean @ConditionalOnMissingBean - public AmazonKinesisAsync amazonKinesis() { - return AmazonKinesisAsyncClientBuilder.standard() - .withCredentials(this.awsCredentialsProvider) - .withRegion(this.region) + public KinesisAsyncClient amazonKinesis() { + return KinesisAsyncClient.builder() + .credentialsProvider(this.awsCredentialsProvider) + .region(this.region) .build(); } @Bean - public KinesisStreamProvisioner provisioningProvider(AmazonKinesisAsync amazonKinesis) { + public KinesisStreamProvisioner provisioningProvider(KinesisAsyncClient amazonKinesis) { return new KinesisStreamProvisioner(amazonKinesis, this.configurationProperties); } @Bean @ConditionalOnMissingBean - public AmazonDynamoDBAsync dynamoDB() { + public DynamoDbAsyncClient dynamoDB() { if (this.hasInputs) { - return AmazonDynamoDBAsyncClientBuilder.standard() - .withCredentials(this.awsCredentialsProvider) - .withRegion(this.region) + return DynamoDbAsyncClient.builder() + .credentialsProvider(this.awsCredentialsProvider) + .region(this.region) .build(); } else { @@ -133,10 +127,10 @@ public class KinesisBinderConfiguration { @Bean @ConditionalOnMissingBean(LockRegistry.class) - @ConditionalOnBean(AmazonDynamoDBAsync.class) + @ConditionalOnBean(DynamoDbAsyncClient.class) @ConditionalOnProperty(name = "spring.cloud.stream.kinesis.binder.kpl-kcl-enabled", havingValue = "false", matchIfMissing = true) - public DynamoDbLockRepository dynamoDbLockRepository(@Autowired(required = false) AmazonDynamoDBAsync dynamoDB) { + public DynamoDbLockRepository dynamoDbLockRepository(@Autowired(required = false) DynamoDbAsyncClient dynamoDB) { if (dynamoDB != null) { KinesisBinderConfigurationProperties.Locks locks = this.configurationProperties.getLocks(); DynamoDbLockRepository dynamoDbLockRepository = new DynamoDbLockRepository(dynamoDB, locks.getTable()); @@ -153,7 +147,7 @@ public class KinesisBinderConfiguration { @Bean @ConditionalOnMissingBean - @ConditionalOnBean(AmazonDynamoDBAsync.class) + @ConditionalOnBean(DynamoDbAsyncClient.class) @ConditionalOnProperty(name = "spring.cloud.stream.kinesis.binder.kpl-kcl-enabled", havingValue = "false", matchIfMissing = true) public LockRegistry dynamoDBLockRegistry( @@ -171,10 +165,10 @@ public class KinesisBinderConfiguration { @Bean @ConditionalOnMissingBean - @ConditionalOnBean(AmazonDynamoDBAsync.class) + @ConditionalOnBean(DynamoDbAsyncClient.class) @ConditionalOnProperty(name = "spring.cloud.stream.kinesis.binder.kpl-kcl-enabled", havingValue = "false", matchIfMissing = true) - public ConcurrentMetadataStore kinesisCheckpointStore(@Autowired(required = false) AmazonDynamoDBAsync dynamoDB) { + public ConcurrentMetadataStore kinesisCheckpointStore(@Autowired(required = false) DynamoDbAsyncClient dynamoDB) { if (dynamoDB != null) { KinesisBinderConfigurationProperties.Checkpoint checkpoint = this.configurationProperties.getCheckpoint(); DynamoDbMetadataStore kinesisCheckpointStore = new DynamoDbMetadataStore(dynamoDB, checkpoint.getTable()); @@ -193,28 +187,14 @@ public class KinesisBinderConfiguration { } } - @Bean - @ConditionalOnMissingBean - public AmazonDynamoDBStreams dynamoDBStreams() { - if (this.hasInputs) { - return AmazonDynamoDBStreamsClientBuilder.standard() - .withCredentials(this.awsCredentialsProvider) - .withRegion(this.region) - .build(); - } - else { - return null; - } - } - @Bean @ConditionalOnMissingBean @ConditionalOnProperty(name = "spring.cloud.stream.kinesis.binder.kpl-kcl-enabled") - public AmazonCloudWatchAsync cloudWatch() { + public CloudWatchAsyncClient cloudWatch() { if (this.hasInputs) { - return AmazonCloudWatchAsyncClientBuilder.standard() - .withCredentials(this.awsCredentialsProvider) - .withRegion(this.region) + return CloudWatchAsyncClient.builder() + .credentialsProvider(this.awsCredentialsProvider) + .region(this.region) .build(); } else { @@ -227,37 +207,35 @@ public class KinesisBinderConfiguration { @ConditionalOnProperty(name = "spring.cloud.stream.kinesis.binder.kpl-kcl-enabled") public KinesisProducerConfiguration kinesisProducerConfiguration() { KinesisProducerConfiguration kinesisProducerConfiguration = new KinesisProducerConfiguration(); - kinesisProducerConfiguration.setCredentialsProvider(this.awsCredentialsProvider); - kinesisProducerConfiguration.setRegion(this.region); + kinesisProducerConfiguration.setCredentialsProvider( + new AWSCredentialsProviderAdapter(this.awsCredentialsProvider)); + kinesisProducerConfiguration.setRegion(this.region.id()); return kinesisProducerConfiguration; } @Bean public KinesisMessageChannelBinder kinesisMessageChannelBinder( KinesisStreamProvisioner provisioningProvider, - AmazonKinesisAsync amazonKinesis, + KinesisAsyncClient amazonKinesis, KinesisExtendedBindingProperties kinesisExtendedBindingProperties, @Autowired(required = false) ConcurrentMetadataStore kinesisCheckpointStore, @Autowired(required = false) LockRegistry lockRegistry, - @Autowired(required = false) AmazonDynamoDB dynamoDBClient, - @Autowired(required = false) AmazonDynamoDBStreams dynamoDBStreams, - @Autowired(required = false) AmazonCloudWatch cloudWatchClient, + @Autowired(required = false) DynamoDbAsyncClient dynamoDBClient, + @Autowired(required = false) CloudWatchAsyncClient cloudWatchClient, @Autowired(required = false) KinesisProducerConfiguration kinesisProducerConfiguration, @Autowired(required = false) ProducerMessageHandlerCustomizer> producerMessageHandlerCustomizer, @Autowired(required = false) ConsumerEndpointCustomizer consumerEndpointCustomizer, - @Autowired List kinesisClientLibConfigurations, @Autowired ObservationRegistry observationRegistry) { KinesisMessageChannelBinder kinesisMessageChannelBinder = new KinesisMessageChannelBinder(this.configurationProperties, provisioningProvider, amazonKinesis, - this.awsCredentialsProvider, dynamoDBClient, dynamoDBStreams, cloudWatchClient); + this.awsCredentialsProvider, dynamoDBClient, cloudWatchClient); kinesisMessageChannelBinder.setCheckpointStore(kinesisCheckpointStore); kinesisMessageChannelBinder.setLockRegistry(lockRegistry); kinesisMessageChannelBinder.setExtendedBindingProperties(kinesisExtendedBindingProperties); kinesisMessageChannelBinder.setKinesisProducerConfiguration(kinesisProducerConfiguration); kinesisMessageChannelBinder.setProducerMessageHandlerCustomizer(producerMessageHandlerCustomizer); kinesisMessageChannelBinder.setConsumerEndpointCustomizer(consumerEndpointCustomizer); - kinesisMessageChannelBinder.setKinesisClientLibConfigurations(kinesisClientLibConfigurations); if (this.configurationProperties.isEnableObservation()) { kinesisMessageChannelBinder.setObservationRegistry(observationRegistry); } @@ -279,4 +257,25 @@ public class KinesisBinderConfiguration { } + private static final class AWSCredentialsProviderAdapter implements AWSCredentialsProvider { + + private final AWSCredentials awsCredentials; + + AWSCredentialsProviderAdapter(AwsCredentialsProvider awsCredentialsProvider) { + AwsCredentials credentials = awsCredentialsProvider.resolveCredentials(); + this.awsCredentials = new BasicAWSCredentials(credentials.secretAccessKey(), credentials.accessKeyId()); + } + + @Override + public AWSCredentials getCredentials() { + return this.awsCredentials; + } + + @Override + public void refresh() { + + } + + } + } diff --git a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/properties/KinesisBinderConfigurationProperties.java b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/properties/KinesisBinderConfigurationProperties.java index 9dae0fd..1f7976a 100644 --- a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/properties/KinesisBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/properties/KinesisBinderConfigurationProperties.java @@ -18,7 +18,7 @@ package org.springframework.cloud.stream.binder.kinesis.properties; import java.time.Duration; -import com.amazonaws.services.dynamodbv2.model.BillingMode; +import software.amazon.awssdk.services.dynamodb.model.BillingMode; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.integration.aws.lock.DynamoDbLockRepository; diff --git a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/properties/KinesisConsumerProperties.java b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/properties/KinesisConsumerProperties.java index 95a0236..965c5c4 100644 --- a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/properties/KinesisConsumerProperties.java +++ b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/properties/KinesisConsumerProperties.java @@ -50,8 +50,6 @@ public class KinesisConsumerProperties { private String shardIteratorType; - private boolean dynamoDbStreams; - private String shardId; /** @@ -134,14 +132,6 @@ public class KinesisConsumerProperties { this.workerId = workerId; } - public boolean isDynamoDbStreams() { - return this.dynamoDbStreams; - } - - public void setDynamoDbStreams(boolean dynamoDbStreams) { - this.dynamoDbStreams = dynamoDbStreams; - } - public String getShardId() { return shardId; } diff --git a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisConsumerDestination.java b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisConsumerDestination.java index 54c53f4..2010778 100644 --- a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisConsumerDestination.java +++ b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisConsumerDestination.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2023 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,7 +18,7 @@ package org.springframework.cloud.stream.binder.kinesis.provisioning; import java.util.List; -import com.amazonaws.services.kinesis.model.Shard; +import software.amazon.awssdk.services.kinesis.model.Shard; import org.springframework.cloud.stream.provisioning.ConsumerDestination; diff --git a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisProducerDestination.java b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisProducerDestination.java index 486c4ac..851ad46 100644 --- a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisProducerDestination.java +++ b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisProducerDestination.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2023 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,7 +18,7 @@ package org.springframework.cloud.stream.binder.kinesis.provisioning; import java.util.List; -import com.amazonaws.services.kinesis.model.Shard; +import software.amazon.awssdk.services.kinesis.model.Shard; import org.springframework.cloud.stream.provisioning.ProducerDestination; diff --git a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisStreamProvisioner.java b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisStreamProvisioner.java index 6a06108..4e7366b 100644 --- a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisStreamProvisioner.java +++ b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisStreamProvisioner.java @@ -16,24 +16,21 @@ package org.springframework.cloud.stream.binder.kinesis.provisioning; -import java.util.ArrayList; -import java.util.Collections; +import java.time.Duration; import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; -import com.amazonaws.services.kinesis.AmazonKinesis; -import com.amazonaws.services.kinesis.model.DescribeStreamRequest; -import com.amazonaws.services.kinesis.model.DescribeStreamResult; -import com.amazonaws.services.kinesis.model.LimitExceededException; -import com.amazonaws.services.kinesis.model.ListShardsRequest; -import com.amazonaws.services.kinesis.model.ListShardsResult; -import com.amazonaws.services.kinesis.model.ResourceNotFoundException; -import com.amazonaws.services.kinesis.model.ScalingType; -import com.amazonaws.services.kinesis.model.Shard; -import com.amazonaws.services.kinesis.model.StreamDescription; -import com.amazonaws.services.kinesis.model.StreamStatus; -import com.amazonaws.services.kinesis.model.UpdateShardCountRequest; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import software.amazon.awssdk.core.retry.backoff.FixedDelayBackoffStrategy; +import software.amazon.awssdk.core.waiters.WaiterResponse; +import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; +import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse; +import software.amazon.awssdk.services.kinesis.model.ListShardsResponse; +import software.amazon.awssdk.services.kinesis.model.ResourceNotFoundException; +import software.amazon.awssdk.services.kinesis.model.ScalingType; +import software.amazon.awssdk.services.kinesis.model.Shard; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; @@ -62,11 +59,11 @@ public class KinesisStreamProvisioner implements private static final Log logger = LogFactory.getLog(KinesisStreamProvisioner.class); - private final AmazonKinesis amazonKinesis; + private final KinesisAsyncClient amazonKinesis; private final KinesisBinderConfigurationProperties configurationProperties; - public KinesisStreamProvisioner(AmazonKinesis amazonKinesis, + public KinesisStreamProvisioner(KinesisAsyncClient amazonKinesis, KinesisBinderConfigurationProperties kinesisBinderConfigurationProperties) { Assert.notNull(amazonKinesis, "'amazonKinesis' must not be null"); @@ -107,13 +104,6 @@ public class KinesisStreamProvisioner implements properties.getHeaderMode() == null || HeaderMode.embeddedHeaders.equals(properties.getHeaderMode())); properties.setHeaderMode(HeaderMode.none); - if (kinesisConsumerProperties.isDynamoDbStreams()) { - if (logger.isInfoEnabled()) { - logger.info("Using DynamoDB table in DynamoDB Streams support for inbound: " + name); - } - return new KinesisConsumerDestination(name, Collections.emptyList()); - } - if (logger.isInfoEnabled()) { logger.info("Using Kinesis stream for inbound: " + name); } @@ -123,111 +113,66 @@ public class KinesisStreamProvisioner implements return new KinesisConsumerDestination(name, createOrUpdate(name, shardCount)); } - private List getShardList(String stream) { - return this.getShardList(stream, 0); - } - - private List getShardList(String stream, int retryCount) { - List shardList = new ArrayList<>(); - - if (retryCount++ > configurationProperties.getDescribeStreamRetries()) { - ResourceNotFoundException resourceNotFoundException = new ResourceNotFoundException( - "The stream [" + stream + "] isn't ACTIVE or doesn't exist."); - resourceNotFoundException.setServiceName("Kinesis"); - - throw new ProvisioningException( - "Kinesis org.springframework.cloud.stream.binder.kinesis.provisioning error", - resourceNotFoundException); - } - - ListShardsRequest listShardsRequest = new ListShardsRequest().withStreamName(stream); - - try { - ListShardsResult listShardsResult = amazonKinesis.listShards(listShardsRequest); - - shardList.addAll(listShardsResult.getShards()); - - } - catch (LimitExceededException limitExceededException) { - logger.info("Got LimitExceededException when describing stream [" + stream + "]. " + "Backing off for [" - + this.configurationProperties.getDescribeStreamBackoff() + "] millis."); - - try { - Thread.sleep(this.configurationProperties.getDescribeStreamBackoff()); - getShardList(stream, retryCount); - } - catch (InterruptedException ex) { - Thread.currentThread().interrupt(); - throw new ProvisioningException( - "The [describeStream] thread for the stream [" + stream + "] has been interrupted.", ex); - } - } - - return shardList; - } - private List createOrUpdate(String stream, int shards) { - List shardList = new ArrayList<>(); + List shardList; try { - shardList = getShardList(stream); + shardList = getShardList(stream).join(); } - catch (ResourceNotFoundException ex) { - if (!this.configurationProperties.isAutoCreateStream()) { + catch (CompletionException ex) { + Throwable cause = ex.getCause(); + if (cause instanceof ResourceNotFoundException) { + if (!this.configurationProperties.isAutoCreateStream()) { + throw new ProvisioningException( + "The stream [" + stream + "] was not found and auto creation is disabled.", cause); + } + if (logger.isInfoEnabled()) { + logger.info("Stream '" + stream + "' not found. Create one..."); + } + + shardList = createStream(stream, shards); + } + else { throw new ProvisioningException( - "The stream [" + stream + "] was not found and auto creation is disabled.", ex); + "Cannot retrieve shards information for stream [" + stream + "].", cause); } - if (logger.isInfoEnabled()) { - logger.info("Stream '" + stream + "' not found. Create one..."); - } - - this.amazonKinesis.createStream(stream, Math.max(this.configurationProperties.getMinShardCount(), shards)); - - waitForStreamToBecomeActive(stream); } int effectiveShardCount = Math.max(this.configurationProperties.getMinShardCount(), shards); - if ((shardList.size() < effectiveShardCount) - && this.configurationProperties.isAutoAddShards()) { + if ((shardList.size() < effectiveShardCount) && this.configurationProperties.isAutoAddShards()) { return updateShardCount(stream, shardList.size(), effectiveShardCount); } return shardList; } - private void waitForStreamToBecomeActive(String streamName) { - int describeStreamRetries = 0; - while (true) { - try { - DescribeStreamResult describeStreamResult = this.amazonKinesis.describeStream(streamName); - if (describeStreamResult != null && - StreamStatus.ACTIVE.name().equals(describeStreamResult.getStreamDescription().getStreamStatus())) { + private CompletableFuture> getShardList(String stream) { + return this.amazonKinesis.describeStreamSummary(request -> request.streamName(stream)) + .thenCompose(reply -> this.amazonKinesis.listShards(request -> request.streamName(stream))) + .thenApply(ListShardsResponse::shards); + } - return; - } - else if (describeStreamRetries++ > this.configurationProperties.getDescribeStreamRetries()) { - ResourceNotFoundException resourceNotFoundException = new ResourceNotFoundException( - "The stream [" + streamName + "] isn't ACTIVE or doesn't exist."); - resourceNotFoundException.setServiceName("Kinesis"); - throw new ProvisioningException( - "Kinesis org.springframework.cloud.stream.binder.kinesis.provisioning error", - resourceNotFoundException); - } - } - catch (LimitExceededException ex) { - logger.info("Got LimitExceededException when describing stream [" + streamName + "]. " + - "Backing off for [" + this.configurationProperties.getDescribeStreamBackoff() + "] millis."); - } - - try { - Thread.sleep(this.configurationProperties.getDescribeStreamBackoff()); - } - catch (InterruptedException ex) { - Thread.currentThread().interrupt(); - throw new ProvisioningException( - "The [describeStream] thread for the stream [" + streamName + "] has been interrupted.", ex); - } + private List createStream(String streamName, int shards) { + try { + return this.amazonKinesis.createStream(request -> request + .streamName(streamName) + .shardCount(Math.max(this.configurationProperties.getMinShardCount(), shards))) + .thenCompose(reply -> waitForStreamToBecomeActive(streamName)) + .thenCompose(reply -> getShardList(streamName)) + .join(); } + catch (Exception ex) { + throw new ProvisioningException("Cannot create stream [" + streamName + "].", ex); + } + } + + private CompletableFuture> waitForStreamToBecomeActive(String streamName) { + return this.amazonKinesis.waiter() + .waitUntilStreamExists(request -> request.streamName(streamName), + waiter -> waiter + .maxAttempts(this.configurationProperties.getDescribeStreamRetries()) + .backoffStrategy(FixedDelayBackoffStrategy.create( + Duration.ofMillis(this.configurationProperties.getDescribeStreamBackoff())))); } private List updateShardCount(String streamName, int shardCount, int targetCount) { @@ -237,66 +182,13 @@ public class KinesisStreamProvisioner implements + "], creating shards..."); } - UpdateShardCountRequest updateShardCountRequest = new UpdateShardCountRequest() - .withStreamName(streamName).withTargetShardCount(targetCount) - .withScalingType(ScalingType.UNIFORM_SCALING); - - this.amazonKinesis.updateShardCount(updateShardCountRequest); - - // Wait for stream to become active again after resharding - List shardList = new ArrayList<>(); - - int describeStreamRetries = 0; - - String exclusiveStartShardId = null; - - DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest() - .withStreamName(streamName); - - while (true) { - DescribeStreamResult describeStreamResult = null; - - try { - describeStreamRequest.withExclusiveStartShardId(exclusiveStartShardId); - describeStreamResult = this.amazonKinesis.describeStream(describeStreamRequest); - StreamDescription streamDescription = describeStreamResult.getStreamDescription(); - if (StreamStatus.ACTIVE.toString().equals(streamDescription.getStreamStatus())) { - shardList.addAll(streamDescription.getShards()); - - if (streamDescription.getHasMoreShards()) { - exclusiveStartShardId = shardList.get(shardList.size() - 1).getShardId(); - } - else { - break; - } - } - } - catch (LimitExceededException ex) { - logger.info("Got LimitExceededException when describing stream [" + streamName + "]. " - + "Backing off for [" + this.configurationProperties.getDescribeStreamBackoff() + "] millis."); - } - - if (describeStreamResult == null || !StreamStatus.ACTIVE.toString().equals( - describeStreamResult.getStreamDescription().getStreamStatus())) { - if (describeStreamRetries++ > this.configurationProperties.getDescribeStreamRetries()) { - ResourceNotFoundException resourceNotFoundException = new ResourceNotFoundException( - "The stream [" + streamName + "] isn't ACTIVE or doesn't exist."); - resourceNotFoundException.setServiceName("Kinesis"); - throw new ProvisioningException( - "Kinesis org.springframework.cloud.stream.binder.kinesis.provisioning error", - resourceNotFoundException); - } - try { - Thread.sleep(this.configurationProperties.getDescribeStreamBackoff()); - } - catch (InterruptedException ex) { - Thread.currentThread().interrupt(); - throw new ProvisioningException( - "The [describeStream] thread for the stream [" + streamName + "] has been interrupted.", ex); - } - } - } - return shardList; + return this.amazonKinesis.updateShardCount(request -> request + .streamName(streamName) + .targetShardCount(targetCount) + .scalingType(ScalingType.UNIFORM_SCALING)) + .thenCompose(reply -> waitForStreamToBecomeActive(streamName)) + .thenCompose(reply -> getShardList(streamName)) + .join(); } } diff --git a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderFunctionalTests.java b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderFunctionalTests.java index 062aaf0..913d6a5 100644 --- a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderFunctionalTests.java +++ b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderFunctionalTests.java @@ -16,7 +16,6 @@ package org.springframework.cloud.stream.binder.kinesis; -import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; @@ -24,14 +23,14 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; -import com.amazonaws.SDKGlobalConfiguration; -import com.amazonaws.services.kinesis.AmazonKinesisAsync; -import com.amazonaws.services.kinesis.model.PutRecordsRequest; -import com.amazonaws.services.kinesis.model.PutRecordsRequestEntry; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; +import software.amazon.awssdk.core.SdkBytes; +import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; +import software.amazon.awssdk.services.kinesis.model.PutRecordsRequest; +import software.amazon.awssdk.services.kinesis.model.PutRecordsRequestEntry; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -63,13 +62,13 @@ import static org.assertj.core.api.Assertions.assertThat; "spring.cloud.stream.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.useNativeDecoding = true", "spring.cloud.stream.kinesis.binder.headers = event.eventType", "spring.cloud.stream.kinesis.binder.autoAddShards = true", - "cloud.aws.region.static=eu-west-2"}) + "spring.cloud.aws.region.static=eu-west-2"}) @DirtiesContext public class KinesisBinderFunctionalTests implements LocalstackContainerTest { static final String KINESIS_STREAM = "test_stream"; - private static AmazonKinesisAsync AMAZON_KINESIS; + private static KinesisAsyncClient AMAZON_KINESIS; @Autowired private ObjectMapper objectMapper; @@ -83,27 +82,31 @@ public class KinesisBinderFunctionalTests implements LocalstackContainerTest { @BeforeAll static void setup() { AMAZON_KINESIS = LocalstackContainerTest.kinesisClient(); - System.setProperty(SDKGlobalConfiguration.AWS_CBOR_DISABLE_SYSTEM_PROPERTY, "true"); } @Test void testKinesisFunction() throws JsonProcessingException, InterruptedException { - PutRecordsRequest putRecordsRequest = new PutRecordsRequest(); - putRecordsRequest.setStreamName(KINESIS_STREAM); + PutRecordsRequest.Builder putRecordsRequest = + PutRecordsRequest.builder() + .streamName(KINESIS_STREAM); + List putRecordsRequestEntryList = new ArrayList<>(); + for (int i = 0; i < 10; i++) { Message eventMessages = MessageBuilder.withPayload("Message" + i) .setHeader("event.eventType", "createEvent") .build(); - PutRecordsRequestEntry putRecordsRequestEntry = new PutRecordsRequestEntry(); - byte[] jsonInput = objectMapper.writeValueAsBytes(eventMessages); - putRecordsRequestEntry.setData(ByteBuffer.wrap(jsonInput)); - putRecordsRequestEntry.setPartitionKey("1"); + PutRecordsRequestEntry putRecordsRequestEntry = + PutRecordsRequestEntry.builder() + .partitionKey("1") + .data(SdkBytes.fromByteArray(objectMapper.writeValueAsBytes(eventMessages))) + .build(); putRecordsRequestEntryList.add(putRecordsRequestEntry); } - putRecordsRequest.setRecords(putRecordsRequestEntryList); - AMAZON_KINESIS.putRecords(putRecordsRequest); + putRecordsRequest.records(putRecordsRequestEntryList); + + AMAZON_KINESIS.putRecords(putRecordsRequest.build()); assertThat(this.messageBarrier.await(30, TimeUnit.SECONDS)).isTrue(); @@ -134,7 +137,7 @@ public class KinesisBinderFunctionalTests implements LocalstackContainerTest { @Bean(destroyMethod = "") - public AmazonKinesisAsync amazonKinesis() { + public KinesisAsyncClient amazonKinesis() { return AMAZON_KINESIS; } diff --git a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderTests.java b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderTests.java index c3ecf2d..e2cbff2 100644 --- a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderTests.java +++ b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderTests.java @@ -16,36 +16,29 @@ package org.springframework.cloud.stream.binder.kinesis; -import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; -import com.amazonaws.SDKGlobalConfiguration; -import com.amazonaws.handlers.AsyncHandler; -import com.amazonaws.services.cloudwatch.AmazonCloudWatch; -import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync; -import com.amazonaws.services.kinesis.AmazonKinesisAsync; -import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream; -import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration; -import com.amazonaws.services.kinesis.model.DescribeStreamRequest; -import com.amazonaws.services.kinesis.model.DescribeStreamResult; -import com.amazonaws.services.kinesis.model.PutRecordRequest; -import com.amazonaws.services.kinesis.model.PutRecordResult; -import com.amazonaws.services.kinesis.model.Record; -import com.amazonaws.services.kinesis.model.Shard; -import com.amazonaws.services.kinesis.model.ShardIteratorType; -import com.amazonaws.services.kinesis.model.StreamDescription; -import com.amazonaws.services.kinesis.model.StreamStatus; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInfo; import org.mockito.BDDMockito; -import org.mockito.stubbing.Answer; +import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient; +import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; +import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; +import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse; +import software.amazon.awssdk.services.kinesis.model.PutRecordRequest; +import software.amazon.awssdk.services.kinesis.model.Record; +import software.amazon.awssdk.services.kinesis.model.Shard; +import software.amazon.awssdk.services.kinesis.model.ShardIteratorType; +import software.amazon.awssdk.services.kinesis.model.StreamStatus; +import software.amazon.kinesis.common.InitialPositionInStream; +import software.amazon.kinesis.common.InitialPositionInStreamExtended; import org.springframework.beans.DirectFieldAccessor; import org.springframework.cloud.stream.binder.Binding; @@ -93,14 +86,13 @@ public class KinesisBinderTests extends ExtendedProducerProperties> implements LocalstackContainerTest { - private static final String CLASS_UNDER_TEST_NAME = KinesisBinderTests.class - .getSimpleName(); + private static final String CLASS_UNDER_TEST_NAME = KinesisBinderTests.class.getSimpleName(); - private static AmazonKinesisAsync AMAZON_KINESIS; + private static KinesisAsyncClient AMAZON_KINESIS; - private static AmazonDynamoDBAsync DYNAMO_DB; + private static DynamoDbAsyncClient DYNAMO_DB; - private static AmazonCloudWatch CLOUD_WATCH; + private static CloudWatchAsyncClient CLOUD_WATCH; public KinesisBinderTests() { @@ -112,7 +104,6 @@ public class KinesisBinderTests extends AMAZON_KINESIS = LocalstackContainerTest.kinesisClient(); DYNAMO_DB = LocalstackContainerTest.dynamoDbClient(); CLOUD_WATCH = LocalstackContainerTest.cloudWatchClient(); - System.setProperty(SDKGlobalConfiguration.AWS_CBOR_DISABLE_SYSTEM_PROPERTY, "true"); } @Test @@ -135,26 +126,25 @@ public class KinesisBinderTests extends consumerProperties.getExtension().setShardIteratorType( ShardIteratorType.AT_TIMESTAMP.name() + ":" + testDate.getTime()); String testStreamName = "nonexisting" + System.currentTimeMillis(); - Binding binding = binder.bindConsumer(testStreamName, "test", output, - consumerProperties); + Binding binding = binder.bindConsumer(testStreamName, "test", output, consumerProperties); binding.unbind(); - DescribeStreamResult streamResult = AMAZON_KINESIS.describeStream(testStreamName); - String createdStreamName = streamResult.getStreamDescription().getStreamName(); - int createdShards = streamResult.getStreamDescription().getShards().size(); - String createdStreamStatus = streamResult.getStreamDescription() - .getStreamStatus(); + DescribeStreamResponse streamResult = + AMAZON_KINESIS.describeStream(request -> request.streamName(testStreamName)) + .join(); + String createdStreamName = streamResult.streamDescription().streamName(); + int createdShards = streamResult.streamDescription().shards().size(); + StreamStatus createdStreamStatus = streamResult.streamDescription().streamStatus(); assertThat(createdStreamName).isEqualTo(testStreamName); - assertThat(createdShards).isEqualTo(consumerProperties.getInstanceCount() - * consumerProperties.getConcurrency()); - assertThat(createdStreamStatus).isEqualTo(StreamStatus.ACTIVE.toString()); + assertThat(createdShards).isEqualTo(consumerProperties.getInstanceCount() * consumerProperties.getConcurrency()); + assertThat(createdStreamStatus).isEqualTo(StreamStatus.ACTIVE); - KinesisShardOffset shardOffset = TestUtils.getPropertyValue(binding, - "lifecycle.streamInitialSequence", KinesisShardOffset.class); + KinesisShardOffset shardOffset = + TestUtils.getPropertyValue(binding, "lifecycle.streamInitialSequence", KinesisShardOffset.class); assertThat(shardOffset.getIteratorType()) .isEqualTo(ShardIteratorType.AT_TIMESTAMP); - assertThat(shardOffset.getTimestamp()).isEqualTo(testDate); + assertThat(Date.from(shardOffset.getTimestamp())).isEqualTo(testDate); } @Test @@ -193,17 +183,17 @@ public class KinesisBinderTests extends .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN) .build()); - Message receivedMessage1 = (Message) receive(input1); + Message receivedMessage1 = (Message) receive(input1, 10); assertThat(receivedMessage1).isNotNull(); assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload1); - Message receivedMessage2 = (Message) receive(input2); + Message receivedMessage2 = (Message) receive(input2, 10); assertThat(receivedMessage2).isNotNull(); assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload1); binding2.unbind(); - String testPayload2 = "foo-" + UUID.randomUUID().toString(); + String testPayload2 = "foo-" + UUID.randomUUID(); output.send(MessageBuilder.withPayload(testPayload2) .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN) .build()); @@ -211,27 +201,27 @@ public class KinesisBinderTests extends binding2 = binder.bindConsumer( String.format("defaultGroup%s0", getDestinationNameDelimiter()), null, input2, consumerProperties); - String testPayload3 = "foo-" + UUID.randomUUID().toString(); + String testPayload3 = "foo-" + UUID.randomUUID(); output.send(MessageBuilder.withPayload(testPayload3) .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN) .build()); - receivedMessage1 = (Message) receive(input1); + receivedMessage1 = (Message) receive(input1, 10); assertThat(receivedMessage1).isNotNull(); assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload2); - receivedMessage1 = (Message) receive(input1); + receivedMessage1 = (Message) receive(input1, 10); assertThat(receivedMessage1).isNotNull(); assertThat(new String(receivedMessage1.getPayload())).isNotNull(); - receivedMessage2 = (Message) receive(input2); + receivedMessage2 = (Message) receive(input2, 10); assertThat(receivedMessage2).isNotNull(); assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload1); - receivedMessage2 = (Message) receive(input2); + receivedMessage2 = (Message) receive(input2, 10); assertThat(receivedMessage2).isNotNull(); assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload2); - receivedMessage2 = (Message) receive(input2); + receivedMessage2 = (Message) receive(input2, 10); assertThat(receivedMessage2).isNotNull(); assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload3); @@ -241,41 +231,36 @@ public class KinesisBinderTests extends } @Test - @SuppressWarnings("unchecked") public void testProducerErrorChannel() throws Exception { KinesisTestBinder binder = getBinder(); - final RuntimeException putRecordException = new RuntimeException( - "putRecordRequestEx"); + final RuntimeException putRecordException = new RuntimeException("putRecordRequestEx"); final AtomicReference sent = new AtomicReference<>(); - AmazonKinesisAsync amazonKinesisMock = mock(AmazonKinesisAsync.class); + KinesisAsyncClient amazonKinesisMock = mock(KinesisAsyncClient.class); BDDMockito - .given(amazonKinesisMock.putRecordAsync(any(PutRecordRequest.class), - any(AsyncHandler.class))) - .willAnswer((Answer>) (invocation) -> { + .given(amazonKinesisMock.putRecord(any(PutRecordRequest.class))) + .willAnswer((invocation) -> { PutRecordRequest request = invocation.getArgument(0); - sent.set(request.getData()); - AsyncHandler handler = invocation.getArgument(1); - handler.onError(putRecordException); - return mock(Future.class); + sent.set(request.data()); + return CompletableFuture.failedFuture(putRecordException); }); - new DirectFieldAccessor(binder.getBinder()).setPropertyValue("amazonKinesis", - amazonKinesisMock); + new DirectFieldAccessor(binder.getBinder()).setPropertyValue("amazonKinesis", amazonKinesisMock); ExtendedProducerProperties producerProps = createProducerProperties(); + producerProps.getExtension().setSync(false); producerProps.setErrorChannelEnabled(true); producerProps.populateBindingName("foobar"); - DirectChannel moduleOutputChannel = createBindableChannel("output", - createProducerBindingProperties(producerProps)); - Binding producerBinding = binder.bindProducer("ec.0", - moduleOutputChannel, producerProps); + DirectChannel moduleOutputChannel = + createBindableChannel("output", createProducerBindingProperties(producerProps)); + Binding producerBinding = + binder.bindProducer("ec.0", moduleOutputChannel, producerProps); - ApplicationContext applicationContext = TestUtils.getPropertyValue( - binder.getBinder(), "applicationContext", ApplicationContext.class); - String s = testBinder.getBinder().getBinderIdentity() + "." + producerProps.getBindingName() + ".errors"; - SubscribableChannel ec = applicationContext.getBean(s, - SubscribableChannel.class); + ApplicationContext applicationContext = + TestUtils.getPropertyValue(binder.getBinder(), "applicationContext", ApplicationContext.class); + String errorChannelName = + testBinder.getBinder().getBinderIdentity() + "." + producerProps.getBindingName() + ".errors"; + SubscribableChannel ec = applicationContext.getBean(errorChannelName, SubscribableChannel.class); final AtomicReference> errorMessage = new AtomicReference<>(); final CountDownLatch latch = new CountDownLatch(1); ec.subscribe((message) -> { @@ -290,11 +275,10 @@ public class KinesisBinderTests extends assertThat(errorMessage.get()).isInstanceOf(ErrorMessage.class); assertThat(errorMessage.get().getPayload()) .isInstanceOf(AwsRequestFailureException.class); - AwsRequestFailureException exception = (AwsRequestFailureException) errorMessage - .get().getPayload(); + AwsRequestFailureException exception = + (AwsRequestFailureException) errorMessage.get().getPayload(); assertThat(exception.getCause()).isSameAs(putRecordException); - assertThat(((PutRecordRequest) exception.getRequest()).getData()) - .isSameAs(sent.get()); + assertThat(((PutRecordRequest) exception.getRequest()).data()).isSameAs(sent.get()); producerBinding.unbind(); } @@ -323,7 +307,7 @@ public class KinesisBinderTests extends Binding inputBinding = binder.bindConsumer("testBatchListener", null, input, consumerProperties); - Message> receivedMessage = (Message>) receive(input); + Message> receivedMessage = (Message>) receive(input, 10); assertThat(receivedMessage).isNotNull(); assertThat(receivedMessage.getPayload().size()).isEqualTo(3); @@ -342,6 +326,7 @@ public class KinesisBinderTests extends KinesisTestBinder binder = getBinder(configurationProperties); DirectChannel output = createBindableChannel("output", new BindingProperties()); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.setAutoStartup(false); Date testDate = new Date(); consumerProperties.getExtension() .setShardIteratorType(ShardIteratorType.AT_TIMESTAMP.name() + ":" + testDate.getTime()); @@ -350,11 +335,10 @@ public class KinesisBinderTests extends Lifecycle lifecycle = TestUtils.getPropertyValue(binding, "lifecycle", Lifecycle.class); assertThat(lifecycle).isInstanceOf(KclMessageDrivenChannelAdapter.class); - KinesisClientLibConfiguration config = - TestUtils.getPropertyValue(lifecycle, "config", KinesisClientLibConfiguration.class); + InitialPositionInStreamExtended initialSequence = + TestUtils.getPropertyValue(lifecycle, "streamInitialSequence", InitialPositionInStreamExtended.class); - assertThat(config.getInitialPositionInStream()).isEqualTo(InitialPositionInStream.AT_TIMESTAMP); - assertThat(config.getTimestampAtInitialPositionInStream()).isEqualTo(testDate); + assertThat(initialSequence).isEqualTo(InitialPositionInStreamExtended.newInitialPositionAtTimestamp(testDate)); binding.unbind(); } @@ -366,15 +350,17 @@ public class KinesisBinderTests extends KinesisTestBinder binder = getBinder(configurationProperties); DirectChannel output = createBindableChannel("output", new BindingProperties()); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.setAutoStartup(false); Binding binding = binder.bindConsumer("testKclStream", null, output, consumerProperties); Lifecycle lifecycle = TestUtils.getPropertyValue(binding, "lifecycle", Lifecycle.class); assertThat(lifecycle).isInstanceOf(KclMessageDrivenChannelAdapter.class); - KinesisClientLibConfiguration config = - TestUtils.getPropertyValue(lifecycle, "config", KinesisClientLibConfiguration.class); + InitialPositionInStreamExtended initialSequence = + TestUtils.getPropertyValue(lifecycle, "streamInitialSequence", InitialPositionInStreamExtended.class); - assertThat(config.getInitialPositionInStream()).isEqualTo(InitialPositionInStream.TRIM_HORIZON); + assertThat(initialSequence) + .isEqualTo(InitialPositionInStreamExtended.newInitialPosition(InitialPositionInStream.TRIM_HORIZON)); binding.unbind(); } @@ -385,7 +371,7 @@ public class KinesisBinderTests extends String stream = "existing" + System.currentTimeMillis(); - AMAZON_KINESIS.createStream(stream, 2); + AMAZON_KINESIS.createStream(request -> request.streamName(stream).shardCount(2)).join(); List shards = describeStream(stream); @@ -406,41 +392,15 @@ public class KinesisBinderTests extends } private List describeStream(String stream) { - String exclusiveStartShardId = null; - - DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest() - .withStreamName(stream); - - List shardList = new ArrayList<>(); - - while (true) { - DescribeStreamResult describeStreamResult; - - describeStreamRequest.withExclusiveStartShardId(exclusiveStartShardId); - describeStreamResult = AMAZON_KINESIS.describeStream(describeStreamRequest); - StreamDescription streamDescription = describeStreamResult - .getStreamDescription(); - if (StreamStatus.ACTIVE.toString() - .equals(streamDescription.getStreamStatus())) { - shardList.addAll(streamDescription.getShards()); - - if (streamDescription.getHasMoreShards()) { - exclusiveStartShardId = shardList.get(shardList.size() - 1) - .getShardId(); - continue; - } - else { - return shardList; - } - } - try { - Thread.sleep(100); - } - catch (InterruptedException ex) { - Thread.currentThread().interrupt(); - throw new IllegalStateException(ex); - } - } + return AMAZON_KINESIS.describeStream(request -> request.streamName(stream)) + .thenCompose(reply -> + AMAZON_KINESIS.waiter().waitUntilStreamExists(request -> request.streamName(stream))) + .join() + .matched() + .response() + .get() + .streamDescription() + .shards(); } @Override @@ -470,8 +430,8 @@ public class KinesisBinderTests extends @Override protected ExtendedConsumerProperties createConsumerProperties() { - ExtendedConsumerProperties kinesisConsumerProperties = new ExtendedConsumerProperties<>( - new KinesisConsumerProperties()); + ExtendedConsumerProperties kinesisConsumerProperties = + new ExtendedConsumerProperties<>(new KinesisConsumerProperties()); // set the default values that would normally be propagated by Spring Cloud Stream kinesisConsumerProperties.setInstanceCount(1); kinesisConsumerProperties.setInstanceIndex(0); @@ -480,7 +440,7 @@ public class KinesisBinderTests extends } private ExtendedProducerProperties createProducerProperties() { - return this.createProducerProperties(null); + return createProducerProperties(null); } @Override diff --git a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisTestBinder.java b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisTestBinder.java index d2d751c..985ca19 100644 --- a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisTestBinder.java +++ b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisTestBinder.java @@ -16,17 +16,13 @@ package org.springframework.cloud.stream.binder.kinesis; -import java.util.ArrayList; -import java.util.List; +import java.util.concurrent.CompletableFuture; -import com.amazonaws.auth.AWSStaticCredentialsProvider; -import com.amazonaws.auth.BasicAWSCredentials; -import com.amazonaws.services.cloudwatch.AmazonCloudWatch; -import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync; -import com.amazonaws.services.kinesis.AmazonKinesisAsync; -import com.amazonaws.services.kinesis.model.ListStreamsRequest; -import com.amazonaws.services.kinesis.model.ListStreamsResult; -import com.amazonaws.services.kinesis.model.ResourceNotFoundException; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient; +import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; +import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; import org.springframework.beans.DirectFieldAccessor; import org.springframework.cloud.stream.binder.AbstractTestBinder; @@ -56,12 +52,13 @@ public class KinesisTestBinder extends AbstractTestBinder, ExtendedProducerProperties> { - private final AmazonKinesisAsync amazonKinesis; + private final KinesisAsyncClient amazonKinesis; private final GenericApplicationContext applicationContext; - public KinesisTestBinder(AmazonKinesisAsync amazonKinesis, AmazonDynamoDBAsync dynamoDbClient, - AmazonCloudWatch cloudWatchClient, KinesisBinderConfigurationProperties kinesisBinderConfigurationProperties) { + public KinesisTestBinder(KinesisAsyncClient amazonKinesis, DynamoDbAsyncClient dynamoDbClient, + CloudWatchAsyncClient cloudWatchClient, + KinesisBinderConfigurationProperties kinesisBinderConfigurationProperties) { this.applicationContext = new AnnotationConfigApplicationContext(Config.class); @@ -75,7 +72,6 @@ public class KinesisTestBinder extends provisioningProvider); binder.setApplicationContext(this.applicationContext); - binder.setKinesisClientLibConfigurations(new ArrayList<>()); setBinder(binder); } @@ -86,39 +82,20 @@ public class KinesisTestBinder extends @Override public void cleanup() { - ListStreamsRequest listStreamsRequest = new ListStreamsRequest(); - ListStreamsResult listStreamsResult = this.amazonKinesis - .listStreams(listStreamsRequest); - - List streamNames = listStreamsResult.getStreamNames(); - - while (listStreamsResult.getHasMoreStreams()) { - if (streamNames.size() > 0) { - listStreamsRequest.setExclusiveStartStreamName( - streamNames.get(streamNames.size() - 1)); - } - listStreamsResult = this.amazonKinesis.listStreams(listStreamsRequest); - streamNames.addAll(listStreamsResult.getStreamNames()); - } - - for (String stream : streamNames) { - this.amazonKinesis.deleteStream(stream); - while (true) { - try { - this.amazonKinesis.describeStream(stream); - try { - Thread.sleep(100); - } - catch (InterruptedException ex) { - Thread.currentThread().interrupt(); - throw new IllegalStateException(ex); - } - } - catch (ResourceNotFoundException ex) { - break; - } - } - } + this.amazonKinesis.listStreams() + .thenCompose(reply -> + CompletableFuture.allOf( + reply.streamNames() + .stream() + .map(streamName -> + this.amazonKinesis.deleteStream(request -> + request.streamName(streamName)) + .thenCompose(result -> + this.amazonKinesis.waiter() + .waitUntilStreamNotExists(request -> + request.streamName(streamName)))) + .toArray(CompletableFuture[]::new))) + .join(); } /** @@ -138,14 +115,15 @@ public class KinesisTestBinder extends private static class TestKinesisMessageChannelBinder extends KinesisMessageChannelBinder { - TestKinesisMessageChannelBinder(AmazonKinesisAsync amazonKinesis, - AmazonDynamoDBAsync dynamoDbClient, - AmazonCloudWatch cloudWatchClient, + TestKinesisMessageChannelBinder(KinesisAsyncClient amazonKinesis, + DynamoDbAsyncClient dynamoDbClient, + CloudWatchAsyncClient cloudWatchClient, KinesisBinderConfigurationProperties kinesisBinderConfigurationProperties, KinesisStreamProvisioner provisioningProvider) { super(kinesisBinderConfigurationProperties, provisioningProvider, amazonKinesis, - new AWSStaticCredentialsProvider(new BasicAWSCredentials("", "")), dynamoDbClient, null, cloudWatchClient); + StaticCredentialsProvider.create(AwsBasicCredentials.create("test", "test")), dynamoDbClient, + cloudWatchClient); } /* diff --git a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/LocalstackContainerTest.java b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/LocalstackContainerTest.java index 1738fb2..5b6990b 100644 --- a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/LocalstackContainerTest.java +++ b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/LocalstackContainerTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2022-2022 the original author or authors. + * Copyright 2022-2023 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,70 +16,73 @@ package org.springframework.cloud.stream.binder.kinesis; -import com.amazonaws.auth.AWSCredentialsProvider; -import com.amazonaws.auth.AWSStaticCredentialsProvider; -import com.amazonaws.auth.BasicAWSCredentials; -import com.amazonaws.client.builder.AwsClientBuilder; -import com.amazonaws.services.cloudwatch.AmazonCloudWatch; -import com.amazonaws.services.cloudwatch.AmazonCloudWatchClientBuilder; -import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync; -import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsyncClientBuilder; -import com.amazonaws.services.kinesis.AmazonKinesisAsync; -import com.amazonaws.services.kinesis.AmazonKinesisAsyncClientBuilder; import org.junit.jupiter.api.BeforeAll; import org.testcontainers.containers.localstack.LocalStackContainer; import org.testcontainers.junit.jupiter.Testcontainers; import org.testcontainers.utility.DockerImageName; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.awscore.client.builder.AwsClientBuilder; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient; +import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; +import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; + /** + * The base contract for JUnit tests based on the container for Localstack. + * The Testcontainers 'reuse' option must be disabled,so, Ryuk container is started + * and will clean all the containers up from this test suite after JVM exit. + * Since the Localstack container instance is shared via static property, it is going to be + * started only once per JVM, therefore the target Docker container is reused automatically. * * @author Artem Bilan - * @author Chris Bono * - * @since 4.0 + * @since 3.0 */ @Testcontainers(disabledWithoutDocker = true) public interface LocalstackContainerTest { + /** + * The shared {@link LocalStackContainer} instance. + */ LocalStackContainer LOCAL_STACK_CONTAINER = - new LocalStackContainer(DockerImageName.parse("localstack/localstack")) - .withServices( - LocalStackContainer.Service.DYNAMODB, - LocalStackContainer.Service.KINESIS, - LocalStackContainer.Service.CLOUDWATCH); + new LocalStackContainer( + DockerImageName.parse("localstack/localstack:1.4.0")) + .withServices( + LocalStackContainer.Service.DYNAMODB, + LocalStackContainer.Service.KINESIS, + LocalStackContainer.Service.CLOUDWATCH); @BeforeAll static void startContainer() { LOCAL_STACK_CONTAINER.start(); } - static AmazonDynamoDBAsync dynamoDbClient() { - return applyAwsClientOptions(AmazonDynamoDBAsyncClientBuilder.standard(), LocalStackContainer.Service.DYNAMODB); + static DynamoDbAsyncClient dynamoDbClient() { + return applyAwsClientOptions(DynamoDbAsyncClient.builder(), LocalStackContainer.Service.DYNAMODB); } - static AmazonKinesisAsync kinesisClient() { - return applyAwsClientOptions(AmazonKinesisAsyncClientBuilder.standard(), LocalStackContainer.Service.KINESIS); + static KinesisAsyncClient kinesisClient() { + return applyAwsClientOptions(KinesisAsyncClient.builder(), LocalStackContainer.Service.KINESIS); } - static AmazonCloudWatch cloudWatchClient() { - return applyAwsClientOptions(AmazonCloudWatchClientBuilder.standard(), LocalStackContainer.Service.CLOUDWATCH); + static CloudWatchAsyncClient cloudWatchClient() { + return applyAwsClientOptions(CloudWatchAsyncClient.builder(), LocalStackContainer.Service.CLOUDWATCH); } - static AWSCredentialsProvider credentialsProvider() { - return new AWSStaticCredentialsProvider( - new BasicAWSCredentials( - LOCAL_STACK_CONTAINER.getAccessKey(), - LOCAL_STACK_CONTAINER.getSecretKey())); + static AwsCredentialsProvider credentialsProvider() { + return StaticCredentialsProvider.create( + AwsBasicCredentials.create(LOCAL_STACK_CONTAINER.getAccessKey(), LOCAL_STACK_CONTAINER.getSecretKey())); } private static , T> T applyAwsClientOptions(B clientBuilder, LocalStackContainer.Service serviceToBuild) { - return clientBuilder.withEndpointConfiguration( - new AwsClientBuilder.EndpointConfiguration( - LOCAL_STACK_CONTAINER.getEndpointOverride(serviceToBuild).toString(), - LOCAL_STACK_CONTAINER.getRegion())) - .withCredentials(credentialsProvider()) + return clientBuilder + .region(Region.of(LOCAL_STACK_CONTAINER.getRegion())) + .credentialsProvider(credentialsProvider()) + .endpointOverride(LOCAL_STACK_CONTAINER.getEndpointOverride(serviceToBuild)) .build(); } - } diff --git a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/adapter/SpringDynamoDBAdapterClientTests.java b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/adapter/SpringDynamoDBAdapterClientTests.java deleted file mode 100644 index 0881123..0000000 --- a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/adapter/SpringDynamoDBAdapterClientTests.java +++ /dev/null @@ -1,200 +0,0 @@ -/* - * Copyright 2017-2019 the original author or authors. - * - * Licensed 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 - * - * https://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.springframework.cloud.stream.binder.kinesis.adapter; - -import java.util.Arrays; -import java.util.Date; -import java.util.List; - -import com.amazonaws.regions.Regions; -import com.amazonaws.services.dynamodbv2.AbstractAmazonDynamoDBStreams; -import com.amazonaws.services.dynamodbv2.AmazonDynamoDBStreams; -import com.amazonaws.services.dynamodbv2.model.AmazonDynamoDBException; -import com.amazonaws.services.dynamodbv2.model.DescribeStreamRequest; -import com.amazonaws.services.dynamodbv2.model.DescribeStreamResult; -import com.amazonaws.services.dynamodbv2.model.KeySchemaElement; -import com.amazonaws.services.dynamodbv2.model.KeyType; -import com.amazonaws.services.dynamodbv2.model.SequenceNumberRange; -import com.amazonaws.services.dynamodbv2.model.Shard; -import com.amazonaws.services.dynamodbv2.model.StreamDescription; -import com.amazonaws.services.dynamodbv2.model.StreamStatus; -import com.amazonaws.services.dynamodbv2.model.StreamViewType; -import com.amazonaws.services.kinesis.model.InvalidArgumentException; -import com.amazonaws.services.kinesis.model.ListShardsRequest; -import com.amazonaws.services.kinesis.model.ListShardsResult; -import com.amazonaws.services.kinesis.model.ResourceNotFoundException; -import org.junit.jupiter.api.Test; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatExceptionOfType; - -/** - * @author Asiel Caballero - * - * @since 2.0.3 - */ -public class SpringDynamoDBAdapterClientTests { - - private final AmazonDynamoDBStreams amazonDynamoDBStreams = new InMemoryAmazonDynamoDBStreams(); - - private final SpringDynamoDBAdapterClient springDynamoDBAdapterClient = - new SpringDynamoDBAdapterClient(amazonDynamoDBStreams); - - @Test - public void listShardsForNotFoundStream() { - assertThatExceptionOfType(ResourceNotFoundException.class) - .isThrownBy(() -> springDynamoDBAdapterClient.listShards(new ListShardsRequest() - .withStreamName("not-found"))); - } - - @Test - public void listShardsWithInvalidToken() { - assertThatExceptionOfType(InvalidArgumentException.class) - .isThrownBy(() -> springDynamoDBAdapterClient.listShards(new ListShardsRequest() - .withNextToken("invalid-token"))); - } - - @Test - public void listShardsWithTokenAndStreamName() { - assertThatExceptionOfType(InvalidArgumentException.class) - .isThrownBy(() -> - springDynamoDBAdapterClient.listShards(new ListShardsRequest() - .withStreamName(InMemoryAmazonDynamoDBStreams.STREAM_ARN) - .withNextToken("valid!!##%%token"))); - } - - @Test - public void listShardsNoPagination() { - ListShardsResult listShards = springDynamoDBAdapterClient.listShards(new ListShardsRequest() - .withStreamName(InMemoryAmazonDynamoDBStreams.STREAM_ARN)); - - assertThat(listShards.getNextToken()).isNull(); - assertThat(listShards.getShards().size()).isEqualTo(InMemoryAmazonDynamoDBStreams.SHARDS.size()); - } - - @Test - public void listShardWithTokenPagination() { - int maxResults = (InMemoryAmazonDynamoDBStreams.SHARDS.size() / 2) - + (InMemoryAmazonDynamoDBStreams.SHARDS.size() % 2); - ListShardsResult listShards = springDynamoDBAdapterClient.listShards(new ListShardsRequest() - .withStreamName(InMemoryAmazonDynamoDBStreams.STREAM_ARN) - .withMaxResults(maxResults)); - - assertThat(listShards.getNextToken()).isNotNull(); - assertThat(listShards.getShards().size()).isEqualTo(maxResults); - - listShards = springDynamoDBAdapterClient.listShards(new ListShardsRequest() - .withNextToken(listShards.getNextToken())); - - assertThat(listShards.getNextToken()).isNull(); - assertThat(listShards.getShards().size()).isEqualTo(InMemoryAmazonDynamoDBStreams.SHARDS.size() - maxResults); - } - - @Test - public void listShardsWithShardIdPagination() { - ListShardsResult listShards = springDynamoDBAdapterClient.listShards(new ListShardsRequest() - .withStreamName(InMemoryAmazonDynamoDBStreams.STREAM_ARN) - .withExclusiveStartShardId(InMemoryAmazonDynamoDBStreams.SHARDS.get(2).getShardId()) - .withMaxResults(1)); - - assertThat(listShards.getNextToken()).isNotNull(); - assertThat(listShards.getShards().size()).isEqualTo(1); - assertThat(listShards.getShards().get(0).getShardId()) - .isEqualTo(InMemoryAmazonDynamoDBStreams.SHARDS.get(3).getShardId()); - } - - - private static class InMemoryAmazonDynamoDBStreams extends AbstractAmazonDynamoDBStreams { - - private static final String TABLE_NAME = "test-streams"; - - private static final String STREAM_LABEL = "2020-10-21T11:49:13.355"; - - private static final String STREAM_ARN = String - .format("arn:aws:dynamodb:%s:%s:table/%s/stream/%s", - Regions.DEFAULT_REGION.getName(), "000000000000", TABLE_NAME, STREAM_LABEL); - - private static final List SHARDS = - Arrays.asList( - buildShard("shardId-00000001603195033866-c5d0c2b1", "51100000000002515059163", "51300000000002521847055"), - buildShard("shardId-00000001603208699318-b15c42af", "804300000000026046960744", "804300000000026046960744") - .withParentShardId("shardId-00000001603195033866-c5d0c2b1"), - buildShard("shardId-00000001603223404428-90b80e6c", "1613900000000033324335703", "1613900000000033324335703") - .withParentShardId("shardId-00000001603208699318-b15c42af"), - buildShard("shardId-00000001603237029376-bd9c40dd", "2364600000000001701561758", "2364600000000001701561758") - .withParentShardId("shardId-00000001603223404428-90b80e6c"), - buildShard("shardId-00000001603249855034-b917a47f", "3071400000000035046301998", "3071400000000035046301998") - .withParentShardId("shardId-00000001603237029376-bd9c40dd")); - - private final StreamDescription streamDescription = new StreamDescription() - .withStreamArn(STREAM_ARN) - .withStreamLabel(STREAM_LABEL) - .withStreamViewType(StreamViewType.KEYS_ONLY) - .withCreationRequestDateTime(new Date()) - .withTableName(TABLE_NAME) - .withKeySchema(new KeySchemaElement("name", KeyType.HASH)) - .withStreamStatus(StreamStatus.ENABLED); - - private static Shard buildShard(String parentShardIdString, String startingSequenceNumber, - String endingSequenceNumber) { - - return new Shard() - .withShardId(parentShardIdString) - .withSequenceNumberRange(new SequenceNumberRange() - .withStartingSequenceNumber(startingSequenceNumber) - .withEndingSequenceNumber(endingSequenceNumber)); - } - - @Override - public DescribeStreamResult describeStream(DescribeStreamRequest request) { - // Invalid StreamArn (Service: AmazonDynamoDBStreams; Status Code: 400; - // Error Code: ValidationException; Request ID: ; Proxy: null) - if (!STREAM_ARN.equals(request.getStreamArn())) { - throw new AmazonDynamoDBException("Invalid StreamArn"); - } - - int limit = request.getLimit() != null ? request.getLimit() : 100; - - int shardIndex = request.getExclusiveStartShardId() != null - ? findShardIndex(request.getExclusiveStartShardId()) - : 0; - int lastShardIndex = Math.min(shardIndex + limit, SHARDS.size()); - streamDescription.setShards(SHARDS.subList(shardIndex, lastShardIndex)); - streamDescription.setLastEvaluatedShardId( - lastShardIndex != SHARDS.size() ? SHARDS.get(lastShardIndex).getShardId() : null); - - return new DescribeStreamResult() - .withStreamDescription(streamDescription); - } - - private int findShardIndex(String shardId) { - int i = 0; - // checkstyle forced this way of writing it - while (i < SHARDS.size() && !SHARDS.get(i).getShardId().equals(shardId)) { - i++; - } - - if (i + 1 >= SHARDS.size()) { - throw new RuntimeException("ShardId not found"); - } - - return i + 1; - } - - } - -} diff --git a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/observation/KinesisBinderObservationTests.java b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/observation/KinesisBinderObservationTests.java index 2cb1322..dc341d8 100644 --- a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/observation/KinesisBinderObservationTests.java +++ b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/observation/KinesisBinderObservationTests.java @@ -25,7 +25,6 @@ import java.util.stream.Collectors; import brave.handler.SpanHandler; import brave.test.TestSpanHandler; -import com.amazonaws.services.kinesis.AmazonKinesisAsync; import io.micrometer.tracing.Span; import io.micrometer.tracing.brave.bridge.BraveFinishedSpan; import io.micrometer.tracing.test.simple.SpansAssert; @@ -34,6 +33,7 @@ import org.apache.commons.logging.LogFactory; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import reactor.core.publisher.Flux; +import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -71,7 +71,7 @@ import static org.awaitility.Awaitility.await; "spring.cloud.stream.kinesis.binder.enable-observation=true", "logging.level.org.springframework.cloud.stream.binder.kinesis.observation=debug", "management.tracing.sampling.probability=1.0", - "cloud.aws.region.static=eu-west-2"}) + "spring.cloud.aws.region.static=eu-west-2"}) @AutoConfigureObservability @DirtiesContext public class KinesisBinderObservationTests implements LocalstackContainerTest { @@ -82,7 +82,7 @@ public class KinesisBinderObservationTests implements LocalstackContainerTest { private static final TestSpanHandler SPANS = new TestSpanHandler(); - private static AmazonKinesisAsync AMAZON_KINESIS; + private static KinesisAsyncClient AMAZON_KINESIS; @Autowired private CountDownLatch messageBarrier; @@ -125,7 +125,7 @@ public class KinesisBinderObservationTests implements LocalstackContainerTest { } @Bean(destroyMethod = "") - public AmazonKinesisAsync amazonKinesis() { + public KinesisAsyncClient amazonKinesis() { return AMAZON_KINESIS; } diff --git a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisStreamProvisionerTests.java b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisStreamProvisionerTests.java index 248e99f..e284eef 100644 --- a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisStreamProvisionerTests.java +++ b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisStreamProvisionerTests.java @@ -16,27 +16,17 @@ package org.springframework.cloud.stream.binder.kinesis.provisioning; -import java.util.Arrays; -import java.util.Collections; -import java.util.List; +import java.time.Duration; -import com.amazonaws.services.kinesis.AmazonKinesis; -import com.amazonaws.services.kinesis.model.CreateStreamResult; -import com.amazonaws.services.kinesis.model.DescribeStreamRequest; -import com.amazonaws.services.kinesis.model.DescribeStreamResult; -import com.amazonaws.services.kinesis.model.ListShardsRequest; -import com.amazonaws.services.kinesis.model.ListShardsResult; -import com.amazonaws.services.kinesis.model.ResourceNotFoundException; -import com.amazonaws.services.kinesis.model.ScalingType; -import com.amazonaws.services.kinesis.model.Shard; -import com.amazonaws.services.kinesis.model.StreamDescription; -import com.amazonaws.services.kinesis.model.StreamStatus; -import com.amazonaws.services.kinesis.model.UpdateShardCountRequest; +import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; -import org.mockito.ArgumentCaptor; +import software.amazon.awssdk.core.retry.backoff.FixedDelayBackoffStrategy; +import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; +import software.amazon.awssdk.services.kinesis.model.ResourceNotFoundException; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.kinesis.LocalstackContainerTest; import org.springframework.cloud.stream.binder.kinesis.properties.KinesisBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kinesis.properties.KinesisConsumerProperties; import org.springframework.cloud.stream.binder.kinesis.properties.KinesisProducerProperties; @@ -46,12 +36,6 @@ import org.springframework.cloud.stream.provisioning.ProvisioningException; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; /** * Tests for the {@link KinesisStreamProvisioner}. @@ -60,255 +44,124 @@ import static org.mockito.Mockito.when; * @author Artem Bilan * @author Sergiu Pantiru */ -class KinesisStreamProvisionerTests { +class KinesisStreamProvisionerTests implements LocalstackContainerTest { + + private static KinesisAsyncClient AMAZON_KINESIS; + + @BeforeAll + static void setup() { + AMAZON_KINESIS = LocalstackContainerTest.kinesisClient(); + } + + private void createStream(String streamName) { + AMAZON_KINESIS.createStream(request -> request.streamName(streamName).shardCount(1)) + .thenCompose(reply -> AMAZON_KINESIS.waiter() + .waitUntilStreamExists(request -> request.streamName(streamName), + waiter -> waiter + .maxAttempts(60) + .backoffStrategy(FixedDelayBackoffStrategy.create(Duration.ofSeconds(1))))) + .join(); + } @Test void testProvisionProducerSuccessfulWithExistingStream() { - AmazonKinesis amazonKinesisMock = mock(AmazonKinesis.class); + String streamName = "provisioning1"; + createStream(streamName); + KinesisBinderConfigurationProperties binderProperties = new KinesisBinderConfigurationProperties(); - KinesisStreamProvisioner provisioner = new KinesisStreamProvisioner( - amazonKinesisMock, binderProperties); + KinesisStreamProvisioner provisioner = new KinesisStreamProvisioner(AMAZON_KINESIS, binderProperties); ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties<>(new KinesisProducerProperties()); - String name = "test-stream"; - when(amazonKinesisMock.listShards(any(ListShardsRequest.class))) - .thenReturn(new ListShardsResult().withShards(new Shard())); - ProducerDestination destination = provisioner.provisionProducerDestination(name, - extendedProducerProperties); + ProducerDestination destination = + provisioner.provisionProducerDestination(streamName, extendedProducerProperties); - verify(amazonKinesisMock).listShards(any(ListShardsRequest.class)); - - assertThat(destination.getName()).isEqualTo(name); + assertThat(destination.getName()).isEqualTo(streamName); + assertThat(destination).isInstanceOf(KinesisProducerDestination.class); + assertThat(destination).extracting("shards").asList().hasSize(1); } @Test void testProvisionConsumerSuccessfulWithExistingStream() { - AmazonKinesis amazonKinesisMock = mock(AmazonKinesis.class); + String streamName = "provisioning2"; + createStream(streamName); + KinesisBinderConfigurationProperties binderProperties = new KinesisBinderConfigurationProperties(); - KinesisStreamProvisioner provisioner = new KinesisStreamProvisioner( - amazonKinesisMock, binderProperties); + KinesisStreamProvisioner provisioner = new KinesisStreamProvisioner(AMAZON_KINESIS, binderProperties); ExtendedConsumerProperties extendedConsumerProperties = - new ExtendedConsumerProperties<>( - new KinesisConsumerProperties()); + new ExtendedConsumerProperties<>(new KinesisConsumerProperties()); - String name = "test-stream"; String group = "test-group"; + ConsumerDestination destination = + provisioner.provisionConsumerDestination(streamName, group, extendedConsumerProperties); - when(amazonKinesisMock.listShards(any(ListShardsRequest.class))) - .thenReturn(new ListShardsResult().withShards(new Shard())); - - ConsumerDestination destination = provisioner.provisionConsumerDestination(name, - group, extendedConsumerProperties); - - verify(amazonKinesisMock).listShards(any(ListShardsRequest.class)); - - assertThat(destination.getName()).isEqualTo(name); + assertThat(destination.getName()).isEqualTo(streamName); + assertThat(destination).isInstanceOf(KinesisConsumerDestination.class); + assertThat(destination).extracting("shards").asList().hasSize(1); } @Test void testProvisionConsumerExistingStreamUpdateShards() { - AmazonKinesis amazonKinesisMock = mock(AmazonKinesis.class); - ArgumentCaptor updateShardCaptor = ArgumentCaptor - .forClass(UpdateShardCountRequest.class); - String name = "test-stream"; - String group = "test-group"; - int targetShardCount = 2; + String streamName = "provisioning3"; + createStream(streamName); + KinesisBinderConfigurationProperties binderProperties = new KinesisBinderConfigurationProperties(); + int targetShardCount = 2; binderProperties.setMinShardCount(targetShardCount); binderProperties.setAutoAddShards(true); - KinesisStreamProvisioner provisioner = new KinesisStreamProvisioner( - amazonKinesisMock, binderProperties); + KinesisStreamProvisioner provisioner = new KinesisStreamProvisioner(AMAZON_KINESIS, binderProperties); ExtendedConsumerProperties extendedConsumerProperties = - new ExtendedConsumerProperties<>( - new KinesisConsumerProperties()); + new ExtendedConsumerProperties<>(new KinesisConsumerProperties()); - DescribeStreamResult describeOriginalStream = describeStreamResultWithShards( - Collections.singletonList(new Shard())); + String group = "test-group"; + ConsumerDestination destination = + provisioner.provisionConsumerDestination(streamName, group, extendedConsumerProperties); - DescribeStreamResult describeUpdatedStream = describeStreamResultWithShards( - Arrays.asList(new Shard(), new Shard())); - - when(amazonKinesisMock.describeStream(any(DescribeStreamRequest.class))) - .thenReturn(describeOriginalStream).thenReturn(describeUpdatedStream); - - when(amazonKinesisMock.listShards(any(ListShardsRequest.class))) - .thenReturn(new ListShardsResult().withShards(new Shard())) - .thenReturn(new ListShardsResult().withShards(new Shard(), new Shard())); - - provisioner.provisionConsumerDestination(name, group, extendedConsumerProperties); - - verify(amazonKinesisMock, times(1)).updateShardCount(updateShardCaptor.capture()); - - assertThat(updateShardCaptor.getValue().getStreamName()).isEqualTo(name); - assertThat(updateShardCaptor.getValue().getScalingType()) - .isEqualTo(ScalingType.UNIFORM_SCALING.name()); - assertThat(updateShardCaptor.getValue().getTargetShardCount()) - .isEqualTo(targetShardCount); + assertThat(destination.getName()).isEqualTo(streamName); + assertThat(destination).isInstanceOf(KinesisConsumerDestination.class); + assertThat(destination).extracting("shards").asList().hasSizeGreaterThanOrEqualTo(2); } @Test void testProvisionProducerSuccessfulWithNewStream() { - AmazonKinesis amazonKinesisMock = mock(AmazonKinesis.class); + String streamName = "provisioning4"; KinesisBinderConfigurationProperties binderProperties = new KinesisBinderConfigurationProperties(); - KinesisStreamProvisioner provisioner = new KinesisStreamProvisioner( - amazonKinesisMock, binderProperties); + KinesisStreamProvisioner provisioner = new KinesisStreamProvisioner(AMAZON_KINESIS, binderProperties); ExtendedProducerProperties extendedProducerProperties = - new ExtendedProducerProperties<>( - new KinesisProducerProperties()); + new ExtendedProducerProperties<>(new KinesisProducerProperties()); - String name = "test-stream"; - Integer shards = 1; + ProducerDestination destination = + provisioner.provisionProducerDestination(streamName, extendedProducerProperties); - when(amazonKinesisMock.listShards(any(ListShardsRequest.class))) - .thenThrow(new ResourceNotFoundException("I got nothing")) - .thenReturn(new ListShardsResult().withShards(new Shard())); - - - when(amazonKinesisMock.createStream(name, shards)) - .thenReturn(new CreateStreamResult()); - - when(amazonKinesisMock.describeStream(name)) - .thenReturn(new DescribeStreamResult() - .withStreamDescription(new StreamDescription() - .withStreamStatus(StreamStatus.ACTIVE))); - - ProducerDestination destination = provisioner.provisionProducerDestination(name, - extendedProducerProperties); - - verify(amazonKinesisMock) - .listShards(any(ListShardsRequest.class)); - - verify(amazonKinesisMock).createStream(name, shards); - - assertThat(destination.getName()).isEqualTo(name); - } - - @Test - void testProvisionProducerUpdateShards() { - AmazonKinesis amazonKinesisMock = mock(AmazonKinesis.class); - ArgumentCaptor updateShardCaptor = ArgumentCaptor - .forClass(UpdateShardCountRequest.class); - String name = "test-stream"; - String group = "test-group"; - int targetShardCount = 2; - KinesisBinderConfigurationProperties binderProperties = new KinesisBinderConfigurationProperties(); - binderProperties.setMinShardCount(targetShardCount); - binderProperties.setAutoAddShards(true); - KinesisStreamProvisioner provisioner = new KinesisStreamProvisioner( - amazonKinesisMock, binderProperties); - - ExtendedConsumerProperties extendedConsumerProperties = - new ExtendedConsumerProperties<>( - new KinesisConsumerProperties()); - - DescribeStreamResult describeOriginalStream = describeStreamResultWithShards( - Collections.singletonList(new Shard())); - - DescribeStreamResult describeUpdatedStream = describeStreamResultWithShards( - Arrays.asList(new Shard(), new Shard())); - - when(amazonKinesisMock.describeStream(any(DescribeStreamRequest.class))) - .thenReturn(describeOriginalStream).thenReturn(describeUpdatedStream); - - when(amazonKinesisMock.listShards(any(ListShardsRequest.class))) - .thenReturn(new ListShardsResult().withShards(new Shard())) - .thenReturn(new ListShardsResult().withShards(new Shard(), new Shard())); - - provisioner.provisionConsumerDestination(name, group, extendedConsumerProperties); - - verify(amazonKinesisMock, times(1)).updateShardCount(updateShardCaptor.capture()); - assertThat(updateShardCaptor.getValue().getStreamName()).isEqualTo(name); - assertThat(updateShardCaptor.getValue().getScalingType()) - .isEqualTo(ScalingType.UNIFORM_SCALING.name()); - assertThat(updateShardCaptor.getValue().getTargetShardCount()) - .isEqualTo(targetShardCount); - } - - @Test - void testProvisionConsumerSuccessfulWithNewStream() { - AmazonKinesis amazonKinesisMock = mock(AmazonKinesis.class); - KinesisBinderConfigurationProperties binderProperties = new KinesisBinderConfigurationProperties(); - KinesisStreamProvisioner provisioner = new KinesisStreamProvisioner( - amazonKinesisMock, binderProperties); - int instanceCount = 1; - int concurrency = 1; - - ExtendedConsumerProperties extendedConsumerProperties = - new ExtendedConsumerProperties<>( - new KinesisConsumerProperties()); - extendedConsumerProperties.setInstanceCount(instanceCount); - extendedConsumerProperties.setConcurrency(concurrency); - - String name = "test-stream"; - String group = "test-group"; - - when(amazonKinesisMock.listShards(any(ListShardsRequest.class))) - .thenThrow(new ResourceNotFoundException("I got nothing")) - .thenReturn(new ListShardsResult().withShards(new Shard())); - - when(amazonKinesisMock.createStream(name, instanceCount * concurrency)) - .thenReturn(new CreateStreamResult()); - - when(amazonKinesisMock.describeStream(name)) - .thenReturn(new DescribeStreamResult() - .withStreamDescription(new StreamDescription() - .withStreamStatus(StreamStatus.ACTIVE))); - - ConsumerDestination destination = provisioner.provisionConsumerDestination(name, - group, extendedConsumerProperties); - - verify(amazonKinesisMock, times(1)) - .listShards(any(ListShardsRequest.class)); - - verify(amazonKinesisMock).createStream(name, instanceCount * concurrency); - - assertThat(destination.getName()).isEqualTo(name); - } - - private static DescribeStreamResult describeStreamResultWithShards( - List shards) { - return new DescribeStreamResult().withStreamDescription(new StreamDescription() - .withShards(shards).withStreamStatus(StreamStatus.ACTIVE) - .withHasMoreShards(Boolean.FALSE)); + assertThat(destination.getName()).isEqualTo(streamName); + assertThat(destination).isInstanceOf(KinesisProducerDestination.class); + assertThat(destination).extracting("shards").asList().hasSize(1); } @Test void testProvisionConsumerResourceNotFoundException() { - AmazonKinesis amazonKinesisMock = mock(AmazonKinesis.class); KinesisBinderConfigurationProperties binderProperties = new KinesisBinderConfigurationProperties(); binderProperties.setAutoCreateStream(false); - KinesisStreamProvisioner provisioner = new KinesisStreamProvisioner( - amazonKinesisMock, binderProperties); + KinesisStreamProvisioner provisioner = new KinesisStreamProvisioner(AMAZON_KINESIS, binderProperties); int instanceCount = 1; int concurrency = 1; ExtendedConsumerProperties extendedConsumerProperties = - new ExtendedConsumerProperties<>( - new KinesisConsumerProperties()); + new ExtendedConsumerProperties<>(new KinesisConsumerProperties()); extendedConsumerProperties.setInstanceCount(instanceCount); extendedConsumerProperties.setConcurrency(concurrency); - String name = "test-stream"; + String name = "provisioning5"; String group = "test-group"; - when(amazonKinesisMock.listShards(any(ListShardsRequest.class))) - .thenThrow(new ResourceNotFoundException("Stream not found")); - assertThatThrownBy(() -> provisioner.provisionConsumerDestination(name, group, extendedConsumerProperties)) .isInstanceOf(ProvisioningException.class) .hasMessageContaining( - "The stream [test-stream] was not found and auto creation is disabled.") + "The stream [provisioning5] was not found and auto creation is disabled.") .hasCauseInstanceOf(ResourceNotFoundException.class); - - verify(amazonKinesisMock, times(1)) - .listShards(any(ListShardsRequest.class)); - - verify(amazonKinesisMock, never()).createStream(name, - instanceCount * concurrency); } }