Add KinesisMessageDrivenChannelAdapter and all the respective infrastructure

* Add mock tests for `KinesisMessageDrivenChannelAdapter`
* Add integration tests based on the Localstack container between
`KinesisMessageHandler` and `KinesisMessageDrivenChannelAdapter`
* Document `KinesisMessageDrivenChannelAdapter`
* Mention via link the `DynamoDbLockRegistry` and `DynamoDbMetadataStore`.
Therefore, add a section id for the Spring Integration in the `dynamodb.adoc`

Author: artembilan

docs/src/main/asciidoc/dynamodb.adoc

@@ -187,6 +187,7 @@ Note that `DynamoDbClientCustomizer` beans are applied **after** `AwsSyncClientC
 Since it depends on how you will use DynamoDb integration providing a list of IAM policies would be pointless since least privilege model should be used.
 To check what IAM policies DynamoDb uses and see which ones you should use please check https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/using-identity-based-policies.html[IAM policies]
 
+[#spring-integration-support]
 === Spring Integration Support
 
 Starting with version 4.0, Spring Cloud AWS provides https://spring.io/projects/spring-integration[Spring Integration] components for Amazon DynamoDB.

docs/src/main/asciidoc/kinesis.adoc

@@ -43,8 +43,79 @@ public MessageHandler kinesisMessageHandler(KinesisAsyncClient amazonKinesis,
 }
 ----
 
-The Kinesis service does not provide a "headers"(attributes) abstraction, so the `KinesisMessageHandler` can be configured with the `OutboundMessageMapper` to embed message headers into the record data alongside the payload.
+The `KinesisMessageDrivenChannelAdapter` is a `MessageProducerSupport` implementation to perform record consumption from the Kinesis stream(s).
+Each shard from the provided streams is managed by an internal state machine to ensure shard records ordering and exclusive shard access from the same consumer group.
+An offset of the records for the consuming shard is stored in the `ConcurrentMetadataStore` via `Checkpointer` abstraction.
+If the `CheckpointMode` of the `KinesisMessageDrivenChannelAdapter` is set to `manual`, then `KinesisHeaders.CHECKPOINTER` is populated to the message this channel adapter produces downstream.
+The check-pointed offset is used to initiate an iterator on the shard after the application restart.
+
+The `KinesisMessageDrivenChannelAdapter` can be configured with a distributed `LockRegistry<?>` to managed exclusive access to the shard in the cluster of the application consuming from the Kinesis.
+For example, the xref:dynamodb.adoc#spring-integration-support[`DynamoDbLockRegistry`] can be used to manage distributed locks via DynamoDB.
+In the case of many instances of the same consuming application in cluster, a distributed implementation of the `ConcurrentMetadataStore` is recommended, too, e.g. xref:dynamodb.adoc#spring-integration-support[`DynamoDbMetadataStore`].
+This way, when one instance leaves the cluster, it releases its locks for shards, and another instance can obtain those locks and continue consuming from the shard according to the stored offset in the mentioned shared meta-data store.
+
+See also `CheckpointMode` Javadocs and related options on the `KinesisMessageDrivenChannelAdapter` for different checkpoint handling algorithms.
+
+The `KinesisShardOffset` abstraction is used to represent an initial shard iterator request for all shards in the provided streams, if there is no specific checkpoint record in the `ConcurrentMetadataStore`.
+Or it can be used as an argument of the overloaded `KinesisMessageDrivenChannelAdapter` to consume from the specific shards.
+The concurrency and checkpoint management remain the same even for an explicit shard configuration.
+
+The `KinesisMessageDrivenChannelAdapter` also supports a `batch` mode to produce a message with a payload as a list of just returned by the shard iterator records.
+Each record data can be converted from `byte[]` via `Converter` setting.
+By default, a `DeserializingConverter` is used based on Java serialization specification.
+Which is aligned with the settings of the mentioned above `KinesisMessageHandler`.
+
+The Kinesis service does not provide a "headers"(attributes) abstraction, so the `KinesisMessageHandler` and `KinesisMessageDrivenChannelAdapter` can be configured with the `OutboundMessageMapper` and `InboundMessageMapper` to embed (and extract) message headers into/from the record data alongside the payload.
 See `EmbeddedHeadersJsonMessageMapper` implementation for more information.
 
+When the shard is closed on the Kinesis service, `KinesisMessageDrivenChannelAdapter` emits a `KinesisShardEndedEvent` into the application context with the key based on the pattern `consumerGroup + ":" + stream + ":" + shardId`.
+Such an event can be useful in any arbitrary application logic where the end of the shard is a crucial indicator, e.g. to perform resharding on the stream.
+
+The following Java Configuration demonstrates some `KinesisMessageDrivenChannelAdapter`:
+
+[source,java]
+----
+@Bean
+public ConcurrentMetadataStore checkpointStore() {
+    return new SimpleMetadataStore();
+}
+
+@Bean
+public LockRegistry lockRegistry() {
+    return new DefaultLockRegistry();
+}
+
+@Bean
+KinesisMessageDrivenChannelAdapter kinesisMessageDrivenChannelAdapter() {
+    var adapter = new KinesisMessageDrivenChannelAdapter(AMAZON_KINESIS_ASYNC, TEST_STREAM);
+    adapter.setOutputChannel(kinesisReceiveChannel());
+    adapter.setErrorChannel(errorChannel());
+    adapter.setErrorMessageStrategy(new KinesisMessageHeaderErrorMessageStrategy());
+    adapter.setCheckpointStore(checkpointStore());
+    adapter.setLockRegistry(lockRegistry());
+    adapter.setEmbeddedHeadersMapper(new EmbeddedHeadersJsonMessageMapper("embedded_header"));
+    adapter.setBindSourceRecord(true);
+    adapter.setDescribeStreamBackoff(10);
+    adapter.setConsumerBackoff(10);
+    adapter.setIdleBetweenPolls(1);
+    return adapter;
+}
+
+@Bean
+public PollableChannel kinesisReceiveChannel() {
+    QueueChannel queueChannel = new QueueChannel();
+    queueChannel.setDatatypes(Date.class);
+    return queueChannel;
+}
+
+@Bean
+public PollableChannel errorChannel() {
+    return new QueueChannel();
+}
+----
+
+
+=== 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-kinesis/src/main/java/io/awspring/cloud/kinesis/integration/CheckpointMode.java

@@ -0,0 +1,48 @@
+/*
+ * 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.integration;
+
+/**
+ * The listener mode, record or batch.
+ *
+ * @author Artem Bilan
+ * @author Hervé Fortin
+ *
+ * @since 4.0
+ */
+public enum CheckpointMode {
+
+	/**
+	 * Checkpoint after each processed record. Makes sense only if {@link ListenerMode#record} is used.
+	 */
+	record,
+
+	/**
+	 * Checkpoint after each processed batch of records.
+	 */
+	batch,
+
+	/**
+	 * Checkpoint on demand via provided to the message {@link Checkpointer} callback.
+	 */
+	manual,
+
+	/**
+	 * Checkpoint at fixed time intervals.
+	 */
+	periodic
+
+}

spring-cloud-aws-kinesis/src/main/java/io/awspring/cloud/kinesis/integration/Checkpointer.java

@@ -0,0 +1,40 @@
+/*
+ * 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.integration;
+
+/**
+ * A callback for target record process to perform checkpoint on the related shard.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+public interface Checkpointer {
+
+	/**
+	 * Checkpoint the currently held sequence number if it is bigger than already stored.
+	 * @return true if checkpoint performed; false otherwise.
+	 */
+	boolean checkpoint();
+
+	/**
+	 * Checkpoint the provided sequence number, if it is bigger than already stored.
+	 * @param sequenceNumber the sequence number to checkpoint.
+	 * @return true if checkpoint performed; false otherwise.
+	 */
+	boolean checkpoint(String sequenceNumber);
+
+}