SpringCloud/spring-cloud-stream-binder-aws-kinesis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-181, GH-182, GH-183: Migrate to AWS SDK v2

Browse Source 
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
This commit is contained in:
abilan
2023-03-24 17:11:19 -04:00
parent 42cd0e50ba
commit fba430a4b0
22 changed files with 525 additions and 1268 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
README.adoc
View File

@@ -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.

16
pom.xml
View File

@@ -15,13 +15,12 @@
<packaging>pom</packaging>
<properties>
<spring-cloud-stream.version>4.0.1</spring-cloud-stream.version>
<java.version>17</java.version>
<spring-cloud-aws.version>2.4.2</spring-cloud-aws.version>
<spring-cloud-stream.version>4.0.1</spring-cloud-stream.version>
<spring-cloud-aws.version>3.0.0-RC1</spring-cloud-aws.version>
<spring-integration-aws.version>3.0.0-SNAPSHOT</spring-integration-aws.version>
<amazon-kinesis-client.version>1.14.9</amazon-kinesis-client.version>
<amazon-kinesis-producer.version>0.14.13</amazon-kinesis-producer.version>
<dynamodb-stream.version>1.5.4</dynamodb-stream.version>
<amazon-kinesis-client.version>2.4.5</amazon-kinesis-client.version>
<amazon-kinesis-producer.version>0.15.5</amazon-kinesis-producer.version>
<testcontainers.version>1.17.6</testcontainers.version>
</properties>
@@ -49,12 +48,7 @@
<version>${spring-integration-aws.version}</version>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>dynamodb-streams-kinesis-adapter</artifactId>
<version>${dynamodb-stream.version}</version>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<groupId>software.amazon.kinesis</groupId>
<artifactId>amazon-kinesis-client</artifactId>
<version>${amazon-kinesis-client.version}</version>
</dependency>

6
spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/building.adoc
View File

@@ -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`.

2
spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/contributing.adoc
View File

@@ -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

76
spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/overview.adoc
View File

@@ -16,11 +16,13 @@ For using the AWS Kinesis Binder, you just need to add it to your Spring Cloud S
</dependency>
----
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.<bindingTarget>.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.<bindingTarget>.consumer.dynamoDbStreams` boolean flag.
In this case a value for `spring.cloud.stream.bindings.<bindingTarget>.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

20
spring-cloud-stream-binder-kinesis/pom.xml
View File

@@ -17,7 +17,7 @@
<dependencies>
<dependency>
<groupId>com.amazonaws</groupId>
<groupId>software.amazon.kinesis</groupId>
<artifactId>amazon-kinesis-client</artifactId>
</dependency>
<dependency>
@@ -34,21 +34,7 @@
</dependency>
<dependency>
<groupId>io.awspring.cloud</groupId>
<artifactId>spring-cloud-starter-aws</artifactId>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-dynamodb</artifactId>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-kinesis</artifactId>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>dynamodb-streams-kinesis-adapter</artifactId>
<artifactId>spring-cloud-aws-starter</artifactId>
</dependency>
<dependency>
@@ -106,7 +92,7 @@
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-tracing-bom</artifactId>
<version>1.1.0-M1</version>
<version>1.1.0-M2</version>
<type>pom</type>
<scope>import</scope>
</dependency>

31
spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderHealthIndicator.java
View File

@@ -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<String> 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();

281
spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java
View File

@@ -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<KinesisClientLibConfiguration> 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<KinesisClientLibConfiguration> 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<KinesisProducerProperties> producerProperties,
MessageChannel errorChannel) {
ExtendedProducerProperties<KinesisProducerProperties> producerProperties, MessageChannel channel,
@Nullable MessageChannel errorChannel) {
FunctionExpression<Message<?>> 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<KinesisProducerProperties> producerProperties, MessageChannel errorChannel) {
return null;
}
private AbstractAwsMessageHandler<?> createKinesisMessageHandler(ProducerDestination destination,
FunctionExpression<Message<?>> partitionKeyExpression, boolean embedHeaders) {
@@ -276,8 +269,7 @@ public class KinesisMessageChannelBinder extends
private AbstractAwsMessageHandler<?> createKplMessageHandler(ProducerDestination destination,
FunctionExpression<Message<?>> 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<KinesisConsumerProperties> 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<KinesisConsumerProperties> 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<KinesisShardOffset> 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]));
}

100
spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/adapter/SpringDynamoDBAdapterClient.java
View File

@@ -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 <a href="https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ListShards.html">ListShards</a>
*/
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;
}
}

125
spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/config/KinesisBinderConfiguration.java
View File

@@ -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<Bindable> 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<? extends AbstractAwsMessageHandler<Void>> producerMessageHandlerCustomizer,
@Autowired(required = false) ConsumerEndpointCustomizer<? extends MessageProducerSupport> consumerEndpointCustomizer,
@Autowired List<KinesisClientLibConfiguration> 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() {
}
}
}

2
spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/properties/KinesisBinderConfigurationProperties.java
View File

@@ -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;

10
spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/properties/KinesisConsumerProperties.java
View File

@@ -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;
}

4
spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisConsumerDestination.java
View File

@@ -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;

4
spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisProducerDestination.java
View File

@@ -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;

234
spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisStreamProvisioner.java
View File

@@ -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<Shard> getShardList(String stream) {
return this.getShardList(stream, 0);
}
private List<Shard> getShardList(String stream, int retryCount) {
List<Shard> 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<Shard> createOrUpdate(String stream, int shards) {
List<Shard> shardList = new ArrayList<>();
List<Shard> 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<List<Shard>> 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<Shard> 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<WaiterResponse<DescribeStreamResponse>> 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<Shard> 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<Shard> 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();
}
}

37
spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderFunctionalTests.java
View File

@@ -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<PutRecordsRequestEntry> putRecordsRequestEntryList = new ArrayList<>();
for (int i = 0; i < 10; i++) {
Message<String> 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;
}

200
spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderTests.java
View File

@@ -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<KinesisProducerProperties>>
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<byte[]> receivedMessage1 = (Message<byte[]>) receive(input1);
Message<byte[]> receivedMessage1 = (Message<byte[]>) receive(input1, 10);
assertThat(receivedMessage1).isNotNull();
assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload1);
Message<byte[]> receivedMessage2 = (Message<byte[]>) receive(input2);
Message<byte[]> receivedMessage2 = (Message<byte[]>) 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<byte[]>) receive(input1);
receivedMessage1 = (Message<byte[]>) receive(input1, 10);
assertThat(receivedMessage1).isNotNull();
assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload2);
receivedMessage1 = (Message<byte[]>) receive(input1);
receivedMessage1 = (Message<byte[]>) receive(input1, 10);
assertThat(receivedMessage1).isNotNull();
assertThat(new String(receivedMessage1.getPayload())).isNotNull();
receivedMessage2 = (Message<byte[]>) receive(input2);
receivedMessage2 = (Message<byte[]>) receive(input2, 10);
assertThat(receivedMessage2).isNotNull();
assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload1);
receivedMessage2 = (Message<byte[]>) receive(input2);
receivedMessage2 = (Message<byte[]>) receive(input2, 10);
assertThat(receivedMessage2).isNotNull();
assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload2);
receivedMessage2 = (Message<byte[]>) receive(input2);
receivedMessage2 = (Message<byte[]>) 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<Object> 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<Future<PutRecordResult>>) (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<KinesisProducerProperties> producerProps = createProducerProperties();
producerProps.getExtension().setSync(false);
producerProps.setErrorChannelEnabled(true);
producerProps.populateBindingName("foobar");
DirectChannel moduleOutputChannel = createBindableChannel("output",
createProducerBindingProperties(producerProps));
Binding<MessageChannel> producerBinding = binder.bindProducer("ec.0",
moduleOutputChannel, producerProps);
DirectChannel moduleOutputChannel =
createBindableChannel("output", createProducerBindingProperties(producerProps));
Binding<MessageChannel> 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<Message<?>> 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<MessageChannel> inputBinding = binder.bindConsumer("testBatchListener",
null, input, consumerProperties);
Message<List<?>> receivedMessage = (Message<List<?>>) receive(input);
Message<List<?>> receivedMessage = (Message<List<?>>) 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<KinesisConsumerProperties> 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<KinesisConsumerProperties> 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<Shard> shards = describeStream(stream);
@@ -406,41 +392,15 @@ public class KinesisBinderTests extends
}
private List<Shard> describeStream(String stream) {
String exclusiveStartShardId = null;
DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest()
.withStreamName(stream);
List<Shard> 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<KinesisConsumerProperties> createConsumerProperties() {
ExtendedConsumerProperties<KinesisConsumerProperties> kinesisConsumerProperties = new ExtendedConsumerProperties<>(
new KinesisConsumerProperties());
ExtendedConsumerProperties<KinesisConsumerProperties> 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<KinesisProducerProperties> createProducerProperties() {
return this.createProducerProperties(null);
return createProducerProperties(null);
}
@Override

80
spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisTestBinder.java
View File

@@ -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<KinesisMessageChannelBinder, ExtendedConsumerProperties<KinesisConsumerProperties>,
ExtendedProducerProperties<KinesisProducerProperties>> {
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<String> 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);
}
/*

73
spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/LocalstackContainerTest.java
View File

@@ -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 <B extends AwsClientBuilder<B, T>, 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();
}
}

200
spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/adapter/SpringDynamoDBAdapterClientTests.java
View File

@@ -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<Shard> 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: <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;
}
}
}

8
spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/observation/KinesisBinderObservationTests.java
View File

@@ -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;
}

281
spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/provisioning/KinesisStreamProvisionerTests.java
View File

@@ -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<KinesisProducerProperties> 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<KinesisConsumerProperties> 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<UpdateShardCountRequest> 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<KinesisConsumerProperties> 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<KinesisProducerProperties> 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<UpdateShardCountRequest> 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<KinesisConsumerProperties> 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<KinesisConsumerProperties> 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<Shard> 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<KinesisConsumerProperties> 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);
}
}