Add support for DynamoDB Streams with Kinesis

* Add `com.amazonaws:dynamodb-streams-kinesis-adapter` dependency which provides
and adapter from DynamoDB Streams Client to Kinesis Client.
That also incudes some convenient API for KCL configuration.
* Add `SpringDynamoDBStreamsAdapterClient` extension to mimic Kinesis API required for the `KinesisMessageDrivenChannelAdapter`
* Support DynamoDB Streams Adapter configuration in the `KclMessageDrivenChannelAdapter`
* Add tests to verify DynamoDB Stream consumption via both `KinesisMessageDrivenChannelAdapter` and `KclMessageDrivenChannelAdapter`
* Improve other KCL tests for race conditions and timing
* Implement DynamoDB Streams support in the Kinesis Binder
* Include `dynamodb-streams-kinesis-adapter` dependency into `spring-cloud-aws-starter-integration-kinesis-client`
since `KclMessageDrivenChannelAdapter` now depends on that adapter API.
* Fix "shard iterator ready" race condition in the `KinesisBinderFunctionalTests`

Related to: spring-cloud/spring-cloud-stream-binder-aws-kinesis#205

Author: artembilan

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

@@ -308,6 +308,11 @@ The KCL graceful shutdown timeout in milliseconds.
 +
 Default: 0 - regular shutdown process
 
+dynamoDbStreams::
+Consume from DynamoDB Streams.
++
+Default: false
+
 The `KclMessageDrivenChannelAdapter` can be customized programmatically for the `ConfigsBuilder` parts.
 For example, to set a custom value for the `LeaseManagementConfig.maxLeasesForWorker` property, the `ConsumerEndpointCustomizer<KclMessageDrivenChannelAdapter>` bean has to be provided:
 
@@ -479,6 +484,15 @@ If [Server-Side Encryption](https://docs.aws.amazon.com/streams/latest/dev/what-
 }
 ----
 
+[[dynamodb-streams-support]]
+=== DynamoDB Streams Support
+
+The Kinesis Binder provides support for consuming CDC events from the https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html[DynamoDB Streams].
+The input binding has to explicitly opt-in for this feature, e.g., `spring.cloud.stream.kinesis.bindings.dynamoDbConsumer-in-0.consumer.dynamoDbStreams=true`.
+The rest of the configuration is done automatically by the binder for both classic and KCL consumers.
+
+NOTE: The binding destination for DynamoDB Stream has to be specified as a Stream ARN on the table to capture changes.
+
 [[telling-the-binder-to-use-your-local-endpoint]]
 === Telling the binder to use your local endpoint
 

docs/src/main/asciidoc/kinesis.adoc

@@ -179,7 +179,17 @@ public KclMessageDrivenChannelAdapter kclMessageDrivenChannelAdapter(KinesisAsyn
 
 NOTE: Unlike `KinesisMessageDrivenChannelAdapter`, the `KclMessageDrivenChannelAdapter` does not support explicit shard assignments.
 
-=== Spring Integration Starters
+==== DynamoDB Streams Support
+
+The `KinesisMessageDrivenChannelAdapter` and `KclMessageDrivenChannelAdapter` provides support for consuming CDC events from the https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html[DynamoDB Streams].
+The `com.amazonaws:dynamodb-streams-kinesis-adapter` dependency must be present on classpath.
+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.
+For consuming via `KinesisMessageDrivenChannelAdapter`, the `SpringDynamoDBStreamsAdapterClient` has to be injected instead of regular `KinesisAsyncClient`.
+The `SpringDynamoDBStreamsAdapterClient` is an extension of the `AmazonDynamoDBStreamsAdapterClient` with overridden `listShards()` and `getRecords()` operations to mimic `KinesisAsyncClient` API called from the `KinesisMessageDrivenChannelAdapter` logic.
+Both channel adapters require a Stream ARN for DynamoDB Stream on the table.
+Using AWS SDK API, such a value is available as a result of the `DescribeTableResponse.table().latestStreamArn()` in the answer to `DynamoDbAsyncClient.describeTable()` request.
+
+==== Spring Integration Starters
 
 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.
 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.

spring-cloud-aws-dependencies/pom.xml

@@ -27,6 +27,7 @@
 		<awssdk-v2.version>2.32.28</awssdk-v2.version>
 		<kcl.version>3.1.2</kcl.version>
 		<kpl.version>1.0.4</kpl.version>
+		<dynamodb-streams.version>2.0.1</dynamodb-streams.version>
 		<amazon.dax.version>2.0.5</amazon.dax.version>
 		<amazon.encryption.s3.version>3.3.5</amazon.encryption.s3.version>
 		<maven-gpg-plugin.version>1.6</maven-gpg-plugin.version>
@@ -294,7 +295,11 @@
 				<artifactId>amazon-kinesis-producer</artifactId>
 				<version>${kpl.version}</version>
 			</dependency>
-
+			<dependency>
+				<groupId>com.amazonaws</groupId>
+				<artifactId>dynamodb-streams-kinesis-adapter</artifactId>
+				<version>${dynamodb-streams.version}</version>
+			</dependency>
 		</dependencies>
 	</dependencyManagement>
 

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

@@ -37,6 +37,10 @@
 			<groupId>software.amazon.kinesis</groupId>
 			<artifactId>amazon-kinesis-producer</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>com.amazonaws</groupId>
+			<artifactId>dynamodb-streams-kinesis-adapter</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-stream</artifactId>

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

@@ -21,6 +21,7 @@
 import io.awspring.cloud.kinesis.integration.KinesisMessageHeaderErrorMessageStrategy;
 import io.awspring.cloud.kinesis.integration.KinesisShardOffset;
 import io.awspring.cloud.kinesis.integration.KplMessageHandler;
+import io.awspring.cloud.kinesis.integration.SpringDynamoDBStreamsAdapterClient;
 import io.awspring.cloud.kinesis.stream.binder.properties.KinesisBinderConfigurationProperties;
 import io.awspring.cloud.kinesis.stream.binder.properties.KinesisConsumerProperties;
 import io.awspring.cloud.kinesis.stream.binder.properties.KinesisExtendedBindingProperties;
@@ -71,6 +72,7 @@
 import org.springframework.util.StringUtils;
 import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient;
 import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
+import software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsClient;
 import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
 import software.amazon.awssdk.services.kinesis.model.InvalidArgumentException;
 import software.amazon.awssdk.services.kinesis.model.Shard;
@@ -112,6 +114,8 @@ public class KinesisMessageChannelBinder extends
 
 	private final BytesMessageMapper embeddedHeadersMapper;
 
+	private final DynamoDbStreamsClient dynamoDBStreams;
+
 	private KinesisExtendedBindingProperties extendedBindingProperties = new KinesisExtendedBindingProperties();
 
 	@Nullable
@@ -130,15 +134,16 @@ public class KinesisMessageChannelBinder extends
 	@SuppressWarnings("removal")
 	public KinesisMessageChannelBinder(KinesisBinderConfigurationProperties configurationProperties,
 			KinesisStreamProvisioner provisioningProvider, KinesisAsyncClient amazonKinesis,
-			@Nullable DynamoDbAsyncClient dynamoDBClient, @Nullable CloudWatchAsyncClient cloudWatchClient) {
+			@Nullable DynamoDbAsyncClient dynamoDBClient, @Nullable DynamoDbStreamsClient dynamoDBStreams,
+			@Nullable CloudWatchAsyncClient cloudWatchClient) {
 
 		super(new String[0], provisioningProvider);
 		Assert.notNull(amazonKinesis, "'amazonKinesis' must not be null");
 		this.configurationProperties = configurationProperties;
 		this.amazonKinesis = amazonKinesis;
 		this.cloudWatchClient = cloudWatchClient;
 		this.dynamoDBClient = dynamoDBClient;
-
+		this.dynamoDBStreams = dynamoDBStreams;
 		this.embeddedHeadersMapper = new org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper(
 				headersToMap(configurationProperties));
 	}
@@ -350,6 +355,9 @@ private MessageProducerSupport createKclConsumerEndpoint(ConsumerDestination des
 
 		KclMessageDrivenChannelAdapter adapter = new KclMessageDrivenChannelAdapter(this.amazonKinesis,
 				this.cloudWatchClient, this.dynamoDBClient, streams);
+		if (kinesisConsumerProperties.isDynamoDbStreams()) {
+			adapter.setDynamoDBStreams(this.dynamoDBStreams);
+		}
 
 		boolean anonymous = !StringUtils.hasText(group);
 		String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID() : group;
@@ -463,24 +471,28 @@ private MessageProducerSupport createKinesisConsumerEndpoint(ConsumerDestination
 			}
 		}
 
+		KinesisAsyncClient amazonKinesisClient = kinesisConsumerProperties.isDynamoDbStreams()
+				? new SpringDynamoDBStreamsAdapterClient(this.dynamoDBStreams)
+				: this.amazonKinesis;
+
 		KinesisMessageDrivenChannelAdapter adapter;
 
 		String shardId = kinesisConsumerProperties.getShardId();
 
 		if (CollectionUtils.isEmpty(shardOffsets) && shardId == null) {
 			String[] streams = Arrays.stream(StringUtils.commaDelimitedListToStringArray(destination.getName()))
 					.map(String::trim).toArray(String[]::new);
-			adapter = new KinesisMessageDrivenChannelAdapter(this.amazonKinesis, streams);
+			adapter = new KinesisMessageDrivenChannelAdapter(amazonKinesisClient, streams);
 		}
 		else if (shardId != null) {
 			Assert.state(!properties.isMultiplex(), "Cannot use multi-stream binding together with 'shard-id'");
 			KinesisShardOffset shardOffset = new KinesisShardOffset(kinesisShardOffset);
 			shardOffset.setStream(destination.getName());
 			shardOffset.setShard(shardId);
-			adapter = new KinesisMessageDrivenChannelAdapter(this.amazonKinesis, shardOffset);
+			adapter = new KinesisMessageDrivenChannelAdapter(amazonKinesisClient, shardOffset);
 		}
 		else {
-			adapter = new KinesisMessageDrivenChannelAdapter(this.amazonKinesis,
+			adapter = new KinesisMessageDrivenChannelAdapter(amazonKinesisClient,
 					shardOffsets.toArray(new KinesisShardOffset[0]));
 		}
 

spring-cloud-aws-kinesis-stream-binder/src/main/java/io/awspring/cloud/kinesis/stream/binder/config/DynamoDbStreamsProperties.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2013-2025 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 io.awspring.cloud.kinesis.stream.binder.config;
+
+import io.awspring.cloud.autoconfigure.AwsClientProperties;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * {@link ConfigurationProperties} for configuring the DynamoDb Streams client.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+@ConfigurationProperties(prefix = "spring.cloud.aws.dynamodb-streams")
+public class DynamoDbStreamsProperties extends AwsClientProperties {
+}

spring-cloud-aws-kinesis-stream-binder/src/main/java/io/awspring/cloud/kinesis/stream/binder/config/KinesisBinderConfiguration.java

@@ -61,6 +61,8 @@
 import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClientBuilder;
 import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
 import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClientBuilder;
+import software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsClient;
+import software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsClientBuilder;
 import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
 import software.amazon.awssdk.services.kinesis.KinesisAsyncClientBuilder;
 import software.amazon.kinesis.producer.KinesisProducerConfiguration;
@@ -79,7 +81,8 @@
 @AutoConfigureAfter({ CredentialsProviderAutoConfiguration.class, RegionProviderAutoConfiguration.class })
 @ConditionalOnMissingBean(Binder.class)
 @EnableConfigurationProperties({ KinesisBinderConfigurationProperties.class, KinesisExtendedBindingProperties.class,
-		KinesisProperties.class, DynamoDbProperties.class, CloudWatchProperties.class })
+		KinesisProperties.class, DynamoDbProperties.class, DynamoDbStreamsProperties.class,
+		CloudWatchProperties.class })
 public class KinesisBinderConfiguration {
 
 	private final KinesisBinderConfigurationProperties configurationProperties;
@@ -216,20 +219,36 @@ public KinesisProducerConfiguration kinesisProducerConfiguration() {
 		return kinesisProducerConfiguration;
 	}
 
+	@Bean
+	@ConditionalOnMissingBean
+	public DynamoDbStreamsClient dynamoDBStreams(DynamoDbStreamsProperties properties,
+			ObjectProvider<AwsClientCustomizer<DynamoDbStreamsClientBuilder>> configurer) {
+		if (this.hasInputs) {
+			return awsClientBuilderConfigurer
+					.configureAsyncClient(DynamoDbStreamsClient.builder(), properties, null, configurer.stream(), null)
+					.build();
+		}
+		else {
+			return null;
+		}
+	}
+
 	@Bean
 	public KinesisMessageChannelBinder kinesisMessageChannelBinder(KinesisStreamProvisioner provisioningProvider,
 			KinesisAsyncClient amazonKinesis, KinesisExtendedBindingProperties kinesisExtendedBindingProperties,
 			@Autowired(required = false) ConcurrentMetadataStore kinesisCheckpointStore,
 			@Autowired(required = false) LockRegistry<?> lockRegistry,
 			@Autowired(required = false) DynamoDbAsyncClient dynamoDBClient,
+			@Autowired(required = false) DynamoDbStreamsClient dynamoDBStreams,
 			@Autowired(required = false) CloudWatchAsyncClient cloudWatchClient,
 			@Autowired(required = false) KinesisProducerConfiguration kinesisProducerConfiguration,
 			@Autowired(required = false) ProducerMessageHandlerCustomizer<? extends AbstractMessageProducingHandler> producerMessageHandlerCustomizer,
 			@Autowired(required = false) ConsumerEndpointCustomizer<? extends MessageProducerSupport> consumerEndpointCustomizer,
 			@Autowired ObservationRegistry observationRegistry) {
 
 		KinesisMessageChannelBinder kinesisMessageChannelBinder = new KinesisMessageChannelBinder(
-				this.configurationProperties, provisioningProvider, amazonKinesis, dynamoDBClient, cloudWatchClient);
+				this.configurationProperties, provisioningProvider, amazonKinesis, dynamoDBClient, dynamoDBStreams,
+				cloudWatchClient);
 		kinesisMessageChannelBinder.setCheckpointStore(kinesisCheckpointStore);
 		kinesisMessageChannelBinder.setLockRegistry(lockRegistry);
 		kinesisMessageChannelBinder.setExtendedBindingProperties(kinesisExtendedBindingProperties);

spring-cloud-aws-kinesis-stream-binder/src/main/java/io/awspring/cloud/kinesis/stream/binder/properties/KinesisConsumerProperties.java

@@ -92,6 +92,8 @@ public class KinesisConsumerProperties {
 
 	private boolean embedHeaders = true;
 
+	private boolean dynamoDbStreams;
+
 	/**
 	 * The {@link MetricsLevel} for emitting (or not) metrics into Cloud Watch.
 	 */
@@ -241,4 +243,12 @@ public void setGracefulShutdownTimeout(long gracefulShutdownTimeout) {
 		this.gracefulShutdownTimeout = gracefulShutdownTimeout;
 	}
 
+	public boolean isDynamoDbStreams() {
+		return this.dynamoDbStreams;
+	}
+
+	public void setDynamoDbStreams(boolean dynamoDbStreams) {
+		this.dynamoDbStreams = dynamoDbStreams;
+	}
+
 }

spring-cloud-aws-kinesis-stream-binder/src/main/java/io/awspring/cloud/kinesis/stream/binder/provisioning/KinesisStreamProvisioner.java

@@ -19,6 +19,7 @@
 import io.awspring.cloud.kinesis.stream.binder.properties.KinesisConsumerProperties;
 import io.awspring.cloud.kinesis.stream.binder.properties.KinesisProducerProperties;
 import java.time.Duration;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
@@ -94,6 +95,11 @@ public ConsumerDestination provisionConsumerDestination(String name, String grou
 			properties.setHeaderMode(HeaderMode.none);
 		}
 
+		if (kinesisConsumerProperties.isDynamoDbStreams()) {
+			logger.info(() -> "Using DynamoDB table in DynamoDB Streams support for inbound: " + name);
+			return new KinesisConsumerDestination(name, Collections.emptyList());
+		}
+
 		int shardCount = properties.getInstanceCount() * properties.getConcurrency();
 
 		if (!properties.isMultiplex()) {

spring-cloud-aws-kinesis-stream-binder/src/test/java/io/awspring/cloud/kinesis/stream/binder/KinesisBinderFunctionalTests.java

@@ -16,6 +16,7 @@
 package io.awspring.cloud.kinesis.stream.binder;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -85,6 +86,23 @@ public class KinesisBinderFunctionalTests implements LocalstackContainerTest {
 	@SuppressWarnings("unchecked")
 	@Test
 	void testKinesisFunction() throws JsonProcessingException, InterruptedException {
+		List<Binding<?>> consumerBindings = this.bindingService
+				.getConsumerBindings("eventConsumerBatchProcessingWithHeaders-in-0");
+
+		assertThat(consumerBindings).hasSize(1);
+
+		Binding<?> binding = consumerBindings.get(0);
+
+		Map<KinesisShardOffset, ?> shardConsumers = TestUtils.getPropertyValue(binding, "lifecycle.shardConsumers",
+				Map.class);
+		assertThat(shardConsumers).hasSize(2).hasKeySatisfying(keySatisfyingCondition(KINESIS_STREAM))
+				.hasKeySatisfying(keySatisfyingCondition("some_other_stream"));
+
+		Object shardConsumer = shardConsumers.values().iterator().next();
+
+		await().untilAsserted(
+				() -> assertThat(TestUtils.getPropertyValue(shardConsumer, "state").toString()).isEqualTo("CONSUME"));
+
 		PutRecordsRequest.Builder putRecordsRequest = PutRecordsRequest.builder().streamName(KINESIS_STREAM);
 
 		List<PutRecordsRequestEntry> putRecordsRequestEntryList = new ArrayList<>();
@@ -118,18 +136,6 @@ void testKinesisFunction() throws JsonProcessingException, InterruptedException
 
 		assertThat(messageFromBatch.getPayload()).isEqualTo("Message0");
 		assertThat(messageFromBatch.getHeaders()).containsEntry("event.eventType", "createEvent");
-
-		List<Binding<?>> consumerBindings = this.bindingService
-				.getConsumerBindings("eventConsumerBatchProcessingWithHeaders-in-0");
-
-		assertThat(consumerBindings).hasSize(1);
-
-		Binding<?> binding = consumerBindings.get(0);
-
-		Map<KinesisShardOffset, ?> shardConsumers = TestUtils.getPropertyValue(binding, "lifecycle.shardConsumers",
-				Map.class);
-		assertThat(shardConsumers).hasSize(2).hasKeySatisfying(keySatisfyingCondition(KINESIS_STREAM))
-				.hasKeySatisfying(keySatisfyingCondition("some_other_stream"));
 	}
 
 	private static Condition<KinesisShardOffset> keySatisfyingCondition(String streamName) {