Skip to content

Commit 1008ffb

Browse files
artembilanartembilan
authored
Add Kinesis Binder for Spring Cloud Stream (#1520)
1 parent 4a4fd90 commit 1008ffb

File tree

43 files changed

+5019
-22
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

43 files changed

+5019
-22
lines changed

‎docs/src/main/asciidoc/images/kinesis-binder.png‎

13.2 KB
Loading

‎docs/src/main/asciidoc/index.adoc‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -160,6 +160,8 @@ include::cloudwatch.adoc[]
160160

161161
include::spring-modulith.adoc[]
162162

163+
include::kinesis-stream-binder.adoc[]
164+
163165
include::testing.adoc[]
164166

165167
include::docker-compose.adoc[]

‎docs/src/main/asciidoc/kinesis-stream-binder.adoc‎

Lines changed: 520 additions & 0 deletions
Large diffs are not rendered by default.

‎docs/src/main/asciidoc/kinesis.adoc‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -179,7 +179,17 @@ public KclMessageDrivenChannelAdapter kclMessageDrivenChannelAdapter(KinesisAsyn
179179

180180
NOTE: Unlike `KinesisMessageDrivenChannelAdapter`, the `KclMessageDrivenChannelAdapter` does not support explicit shard assignments.
181181

182-
=== Spring Integration Starters
182+
==== DynamoDB Streams Support
183+
184+
The `KinesisMessageDrivenChannelAdapter` and `KclMessageDrivenChannelAdapter` provide support for consuming CDC events from the https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html[DynamoDB Streams].
185+
The `com.amazonaws:dynamodb-streams-kinesis-adapter` dependency must be present on classpath.
186+
The KCL provides native support for the mentioned adapter, and only instance of `DynamoDbStreamsClient` has to be injected into the `KclMessageDrivenChannelAdapter` to switch consuming logic from the Kinesis stream to DynamoDB stream.
187+
For consuming via `KinesisMessageDrivenChannelAdapter`, the `SpringDynamoDBStreamsAdapterClient` has to be injected instead of regular `KinesisAsyncClient`.
188+
The `SpringDynamoDBStreamsAdapterClient` is an extension of the `AmazonDynamoDBStreamsAdapterClient` with overridden `listShards()` and `getRecords()` operations to mimic `KinesisAsyncClient` API called from the `KinesisMessageDrivenChannelAdapter` logic.
189+
Both channel adapters require a Stream ARN for DynamoDB Stream on the table.
190+
Using AWS SDK API, such a Stream ARN value is available as a result of the `DescribeTableResponse.table().latestStreamArn()` in the answer to `DynamoDbAsyncClient.describeTable()` request.
191+
192+
==== Spring Integration Starters
183193

184194
The Spring Integration dependency in the `spring-cloud-aws-kinesis` module is `optional` to avoid unnecessary artifacts on classpath when Spring Integration is not used.
185195
For convenience, a dedicated `spring-cloud-aws-starter-integration-kinesis` is provided managing all the required dependencies for Spring Integration support with a classical Amazon Kinesis client.

‎pom.xml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@
4545
<module>spring-cloud-aws-sqs</module>
4646
<module>spring-cloud-aws-dynamodb</module>
4747
<module>spring-cloud-aws-kinesis</module>
48+
<module>spring-cloud-aws-kinesis-stream-binder</module>
4849
<module>spring-cloud-aws-starters/spring-cloud-aws-starter-integration-kinesis</module>
4950
<module>spring-cloud-aws-starters/spring-cloud-aws-starter-integration-kinesis-producer</module>
5051
<module>spring-cloud-aws-starters/spring-cloud-aws-starter-integration-kinesis-client</module>

‎spring-cloud-aws-dependencies/pom.xml‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,9 +28,11 @@
2828
<kcl.version>3.1.2</kcl.version>
2929
<kpl.version>1.0.4</kpl.version>
3030
<amazon.dax.version>2.0.6</amazon.dax.version>
31+
<dynamodb-streams.version>2.0.1</dynamodb-streams.version>
3132
<amazon.encryption.s3.version>3.3.5</amazon.encryption.s3.version>
3233
<maven-gpg-plugin.version>1.6</maven-gpg-plugin.version>
3334
<spring-cloud-commons.version>5.0.0-RC1</spring-cloud-commons.version>
35+
<spring-cloud-stream.version>5.0.0</spring-cloud-stream.version>
3436
<jakarta.mail.version>2.1.3</jakarta.mail.version>
3537
<eclipse.jakarta.mail.version>2.0.3</eclipse.jakarta.mail.version>
3638
<spring-modulith.version>2.0.0-RC1</spring-modulith.version>
@@ -54,6 +56,13 @@
5456
<type>pom</type>
5557
<scope>import</scope>
5658
</dependency>
59+
<dependency>
60+
<groupId>org.springframework.cloud</groupId>
61+
<artifactId>spring-cloud-stream-dependencies</artifactId>
62+
<version>${spring-cloud-stream.version}</version>
63+
<type>pom</type>
64+
<scope>import</scope>
65+
</dependency>
5766

5867
<dependency>
5968
<groupId>org.springframework.modulith</groupId>
@@ -118,6 +127,11 @@
118127
<artifactId>spring-cloud-aws-kinesis</artifactId>
119128
<version>${project.version}</version>
120129
</dependency>
130+
<dependency>
131+
<groupId>io.awspring.cloud</groupId>
132+
<artifactId>spring-cloud-aws-kinesis-stream-binder</artifactId>
133+
<version>${project.version}</version>
134+
</dependency>
121135
<dependency>
122136
<groupId>io.awspring.cloud</groupId>
123137
<artifactId>spring-cloud-aws-starter-integration-kinesis</artifactId>
@@ -281,7 +295,11 @@
281295
<artifactId>amazon-kinesis-producer</artifactId>
282296
<version>${kpl.version}</version>
283297
</dependency>
284-
298+
<dependency>
299+
<groupId>com.amazonaws</groupId>
300+
<artifactId>dynamodb-streams-kinesis-adapter</artifactId>
301+
<version>${dynamodb-streams.version}</version>
302+
</dependency>
285303
</dependencies>
286304
</dependencyManagement>
287305

‎spring-cloud-aws-kinesis-stream-binder/pom.xml‎

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<project xmlns="http://maven.apache.org/POM/4.0.0"
3+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5+
<modelVersion>4.0.0</modelVersion>
6+
<parent>
7+
<groupId>io.awspring.cloud</groupId>
8+
<artifactId>spring-cloud-aws</artifactId>
9+
<version>4.0.0-M1</version>
10+
</parent>
11+
12+
<artifactId>spring-cloud-aws-kinesis-stream-binder</artifactId>
13+
<name>Spring Cloud Stream AWS Kinesis Binder</name>
14+
15+
<dependencies>
16+
<dependency>
17+
<groupId>io.awspring.cloud</groupId>
18+
<artifactId>spring-cloud-aws-starter</artifactId>
19+
</dependency>
20+
<dependency>
21+
<groupId>io.awspring.cloud</groupId>
22+
<artifactId>spring-cloud-aws-kinesis</artifactId>
23+
</dependency>
24+
<dependency>
25+
<groupId>io.awspring.cloud</groupId>
26+
<artifactId>spring-cloud-aws-dynamodb</artifactId>
27+
</dependency>
28+
<dependency>
29+
<groupId>software.amazon.awssdk</groupId>
30+
<artifactId>kinesis</artifactId>
31+
</dependency>
32+
<dependency>
33+
<groupId>software.amazon.kinesis</groupId>
34+
<artifactId>amazon-kinesis-client</artifactId>
35+
</dependency>
36+
<dependency>
37+
<groupId>software.amazon.kinesis</groupId>
38+
<artifactId>amazon-kinesis-producer</artifactId>
39+
</dependency>
40+
<dependency>
41+
<groupId>com.amazonaws</groupId>
42+
<artifactId>dynamodb-streams-kinesis-adapter</artifactId>
43+
</dependency>
44+
<dependency>
45+
<groupId>org.springframework.cloud</groupId>
46+
<artifactId>spring-cloud-stream</artifactId>
47+
</dependency>
48+
49+
<dependency>
50+
<groupId>org.springframework.boot</groupId>
51+
<artifactId>spring-boot-configuration-processor</artifactId>
52+
<optional>true</optional>
53+
</dependency>
54+
<dependency>
55+
<groupId>org.springframework.boot</groupId>
56+
<artifactId>spring-boot-starter-actuator</artifactId>
57+
<optional>true</optional>
58+
</dependency>
59+
60+
<dependency>
61+
<groupId>org.springframework.cloud</groupId>
62+
<artifactId>spring-cloud-stream-test-support</artifactId>
63+
<scope>test</scope>
64+
</dependency>
65+
<dependency>
66+
<groupId>org.springframework.boot</groupId>
67+
<artifactId>spring-boot-micrometer-tracing-brave</artifactId>
68+
<scope>test</scope>
69+
</dependency>
70+
<dependency>
71+
<groupId>org.springframework.boot</groupId>
72+
<artifactId>spring-boot-micrometer-tracing-test</artifactId>
73+
<scope>test</scope>
74+
</dependency>
75+
<dependency>
76+
<groupId>io.micrometer</groupId>
77+
<artifactId>micrometer-tracing-integration-test</artifactId>
78+
<scope>test</scope>
79+
<exclusions>
80+
<exclusion>
81+
<groupId>io.opentelemetry</groupId>
82+
<artifactId>*</artifactId>
83+
</exclusion>
84+
<exclusion>
85+
<groupId>com.wavefront</groupId>
86+
<artifactId>*</artifactId>
87+
</exclusion>
88+
<exclusion>
89+
<groupId>io.micrometer</groupId>
90+
<artifactId>micrometer-tracing-bridge-otel</artifactId>
91+
</exclusion>
92+
</exclusions>
93+
</dependency>
94+
<dependency>
95+
<groupId>org.testcontainers</groupId>
96+
<artifactId>testcontainers-localstack</artifactId>
97+
<scope>test</scope>
98+
</dependency>
99+
<dependency>
100+
<groupId>org.testcontainers</groupId>
101+
<artifactId>testcontainers-junit-jupiter</artifactId>
102+
<scope>test</scope>
103+
</dependency>
104+
</dependencies>
105+
106+
</project>

‎spring-cloud-aws-kinesis-stream-binder/src/main/java/io/awspring/cloud/kinesis/stream/binder/KinesisBinderHealthIndicator.java‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
/*
2+
* Copyright 2013-2025 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.awspring.cloud.kinesis.stream.binder;
17+
18+
import java.util.ArrayList;
19+
import java.util.List;
20+
import java.util.concurrent.CompletionException;
21+
import java.util.concurrent.TimeUnit;
22+
import org.springframework.boot.health.contributor.Health;
23+
import org.springframework.boot.health.contributor.HealthIndicator;
24+
import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
25+
import software.amazon.awssdk.services.kinesis.model.LimitExceededException;
26+
27+
/**
28+
* @author Artem Bilan
29+
*
30+
* @since 4.0
31+
*/
32+
public class KinesisBinderHealthIndicator implements HealthIndicator {
33+
34+
private final KinesisMessageChannelBinder kinesisMessageChannelBinder;
35+
36+
public KinesisBinderHealthIndicator(KinesisMessageChannelBinder kinesisMessageChannelBinder) {
37+
this.kinesisMessageChannelBinder = kinesisMessageChannelBinder;
38+
}
39+
40+
@Override
41+
public Health health() {
42+
KinesisAsyncClient amazonKinesis = this.kinesisMessageChannelBinder.getAmazonKinesis();
43+
List<String> streamsInUse = new ArrayList<>(this.kinesisMessageChannelBinder.getStreamsInUse());
44+
for (String stream : streamsInUse) {
45+
while (true) {
46+
try {
47+
amazonKinesis.listShards(request -> request.streamName(stream).maxResults(1)).join();
48+
break;
49+
}
50+
catch (CompletionException ex) {
51+
Throwable cause = ex.getCause();
52+
if (cause instanceof LimitExceededException) {
53+
try {
54+
TimeUnit.SECONDS.sleep(1);
55+
}
56+
catch (InterruptedException e) {
57+
Thread.currentThread().interrupt();
58+
return Health.down().withException(ex).build();
59+
}
60+
}
61+
else {
62+
return Health.down().withException(ex).build();
63+
}
64+
}
65+
}
66+
}
67+
return Health.up().build();
68+
}
69+
70+
}

0 commit comments

Comments
 (0)