Merge branch 'main' into 1457-kinesis-autoconfiguration

# Conflicts:
#	pom.xml

Author: MatejNedic

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/index.adoc

@@ -140,6 +140,8 @@ include::core.adoc[]
 
 include::dynamodb.adoc[]
 
+include::kinesis.adoc[]
+
 include::s3.adoc[]
 
 include::ses.adoc[]

docs/src/main/asciidoc/kinesis.adoc

@@ -0,0 +1,121 @@
+[#spring-cloud-aws-kinesis]
+== Kinesis Integration
+
+The https://aws.amazon.com/kinesis/[Kinesis] is a platform for streaming data on AWS, making it easy to load and analyze streaming data and also providing the ability for you to build custom streaming data applications for specialized needs.
+
+// TODO: auto-configuration
+
+=== Spring Integration Support
+
+Also, starting with version 4.0, Spring Cloud AWS provides https://spring.io/projects/spring-integration[Spring Integration] channel adapters for Amazon Kinesis.
+
+The `KinesisMessageHandler` is an `AbstractMessageHandler` to perform put record(s) to the Kinesis stream.
+The stream, partition key (or explicit hash key) and sequence number can be determined against a request message via evaluation provided expressions or can be specified statically.
+They also can be specified as `KinesisHeaders.STREAM`, `KinesisHeaders.PARTITION_KEY` and `KinesisHeaders.SEQUENCE_NUMBER` respectively.
+
+The `KinesisMessageHandler` can be configured with the `outputChannel` for sending a `Message` on successful put operation.
+The payload is the original request and additional `KinesisHeaders.SHARD` and `KinesisHeaders.SEQUENCE_NUMBER` headers are populated from the `PutRecordResposne`.
+If the request payload is a `PutRecordsRequest`, the full `PutRecordsResponse` is populated in the `KinesisHeaders.SERVICE_RESULT` header instead.
+
+When an async failure is happened on the put operation, the `ErrorMessage` is sent to the `errorChannel` header or global one.
+The payload is an `MessageHandlingException`.
+
+The `payload` of request message can be:
+
+- `PutRecordsRequest` to perform `KinesisAsyncClient.putRecords`
+- `PutRecordRequest` to perform `KinesisAsyncClient.putRecord`
+- `ByteBuffer` to represent data of the `PutRecordRequest`
+- `byte[]` which is wrapped to the `ByteBuffer`
+- any other type that is converted to the `byte[]` by the provided `Converter`; the `SerializingConverter` is used by default.
+
+The Java Configuration for the message handler is:
+
+[source,java]
+----
+@Bean
+@ServiceActivator(inputChannel = "kinesisSendChannel")
+public MessageHandler kinesisMessageHandler(KinesisAsyncClient amazonKinesis,
+                                            MessageChannel channel) {
+    KinesisMessageHandler kinesisMessageHandler = new KinesisMessageHandler(amazonKinesis);
+    kinesisMessageHandler.setPartitionKey("1");
+    kinesisMessageHandler.setOutputChannel(channel);
+    return kinesisMessageHandler;
+}
+----
+
+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.

docs/src/main/asciidoc/s3.adoc

@@ -604,17 +604,51 @@ The Spring Boot Starter for S3 provides the following configuration options:
 | `spring.cloud.aws.s3.config.reload.max-wait-time-for-restart` | `Duration`| `2s` | The maximum time between the detection of changes in property source and the application context restart when `restart_context` strategy is used.
 |===
 
+=== S3 Vector Client support
+
+https://aws.amazon.com/blogs/aws/introducing-amazon-s3-vectors-first-cloud-storage-with-native-vector-support-at-scale/[S3 Vector Store] is a vector storage which supports uploading, storing and querying vectors.
+To allow users simpler use of `S3VectorsClient` with https://github.com/spring-projects/spring-ai[Spring AI] project or use of a plain client, we support autoconfiguration of the `S3VectorsClient`.
+
+To enable autoconfiguration you should add the following dependencies
+[source,xml]
+----
+<dependency>
+    <groupId>io.awspring.cloud</groupId>
+    <artifactId>spring-cloud-aws-starter</artifactId>
+</dependency>
+
+<dependency>
+	<groupId>software.amazon.awssdk</groupId>
+	<artifactId>s3vectors</artifactId>
+</dependency>
+----
+
+After dependencies are introduced Spring Cloud AWS will automatically create a `S3VectorsClient` bean which can than be autowired and used.
+
+[cols="2,3,1,1"]
+|===
+| Name | Description | Required | Default value
+| `spring.cloud.aws.s3.vector.config.enabled` | Enables the S3 config import integration. | No | `true`
+| `spring.cloud.aws.s3.config.reload.strategy` | `Enum` | `refresh` | The strategy to use when firing a reload (`refresh`, `restart_context`)
+| `spring.cloud.aws.s3.config.reload.period` | `Duration`| `15s` | The period for verifying changes
+| `spring.cloud.aws.s3.config.reload.max-wait-time-for-restart` | `Duration`| `2s` | The maximum time between the detection of changes in property source and the application context restart when `restart_context` strategy is used.
+|===
 
 === IAM Permissions
 
 Following IAM permissions are required by Spring Cloud AWS:
 
-[cols="2,1"]
-|===
-| Downloading files | `s3:GetObject`
-| Searching files | `s3:ListObjects`
-| Uploading files | `s3:PutObject`
+[cols="2,3,1,1"]
 |===
+| Name | Description | Required | Default value
+| `spring.cloud.aws.s3.vector.enabled` | Enables the S3VectorsClient autoconfiguration. | No | `true`
+| `spring.cloud.aws.s3.vector.endpoint` | Configures endpoint used by `S3VectorsClient`. | No | `http://localhost:4566`
+| `spring.cloud.aws.s3.vector.region` | Configures region used by `S3VectorsClient`. | No | `eu-west-1`
+
+
+
+
+=== Example of IAM policy for Spring Cloud AWS demo bucket
 
 Sample IAM policy granting access to `spring-cloud-aws-demo` bucket:
 
@@ -641,3 +675,149 @@ Sample IAM policy granting access to `spring-cloud-aws-demo` bucket:
     ]
 }
 ----
+
+=== Spring Integration Support
+
+Starting with version 4.0, Spring Cloud AWS provides https://spring.io/projects/spring-integration[Spring Integration] channel adapters for Amazon SQS.
+
+The S3 Channel Adapters are based on the `S3Client` template and `S3TransferManager`.
+See their specification and Javadocs for more information.
+
+The S3 Inbound Channel Adapter is represented by the `S3InboundFileSynchronizingMessageSource` and allows pulling S3 objects as files from the S3 bucket to the local directory for synchronization.
+This adapter is fully similar to the Inbound Channel Adapters in the FTP and SFTP Spring Integration modules.
+See more information in the https://docs.spring.io/spring-integration/reference/ftp.html[FTP/FTPS Adapters Chapter] for common options or `SessionFactory`, `RemoteFileTemplate` and `FileListFilter` abstractions.
+
+The Java Configuration is:
+
+[source,java]
+----
+@SpringBootApplication
+public static class MyConfiguration {
+
+    @Autowired
+    private S3Client amazonS3;
+
+    @Bean
+    public S3InboundFileSynchronizer s3InboundFileSynchronizer() {
+    	S3InboundFileSynchronizer synchronizer = new S3InboundFileSynchronizer(this.amazonS3);
+    	synchronizer.setDeleteRemoteFiles(true);
+    	synchronizer.setPreserveTimestamp(true);
+    	synchronizer.setRemoteDirectory(S3_BUCKET);
+    	synchronizer.setFilter(new S3RegexPatternFileListFilter(".*\\.test$"));
+    	Expression expression = PARSER.parseExpression("#this.toUpperCase() + '.a'");
+    	synchronizer.setLocalFilenameGeneratorExpression(expression);
+    	return synchronizer;
+    }
+
+    @Bean
+    @InboundChannelAdapter(value = "s3FilesChannel", poller = @Poller(fixedDelay = "100"))
+    public S3InboundFileSynchronizingMessageSource s3InboundFileSynchronizingMessageSource() {
+    	S3InboundFileSynchronizingMessageSource messageSource =
+    			new S3InboundFileSynchronizingMessageSource(s3InboundFileSynchronizer());
+    	messageSource.setAutoCreateLocalDirectory(true);
+    	messageSource.setLocalDirectory(LOCAL_FOLDER);
+    	messageSource.setLocalFilter(new AcceptOnceFileListFilter<File>());
+    	return messageSource;
+    }
+
+    @Bean
+    public PollableChannel s3FilesChannel() {
+    	return new QueueChannel();
+    }
+}
+----
+
+With this config you receive messages with `java.io.File` `payload` from the `s3FilesChannel` after periodic synchronization of content from the Amazon S3 bucket into the local directory.
+
+The `S3StreamingMessageSource` adapter produces messages with payloads of type `InputStream`, allowing S3 objects to be fetched without writing to the local file system.
+Since the session remains open, the consuming application is responsible for closing the session when the file has been consumed.
+The session is provided in the closeableResource header (`IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE`).
+Standard framework components, such as the `FileSplitter` and `StreamTransformer` will automatically close the session.
+
+The following Spring Boot application provides an example of configuring the S3 inbound streaming adapter using Java configuration:
+
+[source,java]
+----
+@SpringBootApplication
+public class S3JavaApplication {
+
+    public static void main(String[] args) {
+        new SpringApplicationBuilder(S3JavaApplication.class)
+            .web(false)
+            .run(args);
+    }
+
+    @Autowired
+    private S3Client amazonS3;
+
+    @Bean
+    @InboundChannelAdapter(value = "s3Channel", poller = @Poller(fixedDelay = "100"))
+    public MessageSource<InputStream> s3InboundStreamingMessageSource() {
+        S3StreamingMessageSource messageSource = new S3StreamingMessageSource(template());
+        messageSource.setRemoteDirectory(S3_BUCKET);
+        messageSource.setFilter(new S3PersistentAcceptOnceFileListFilter(new SimpleMetadataStore(),
+                                   "streaming"));
+    	return messageSource;
+    }
+
+    @Bean
+    @Transformer(inputChannel = "s3Channel", outputChannel = "data")
+    public org.springframework.integration.transformer.Transformer transformer() {
+        return new StreamTransformer();
+    }
+
+    @Bean
+    public S3RemoteFileTemplate template() {
+        return new S3RemoteFileTemplate(new S3SessionFactory(this.amazonS3));
+    }
+
+    @Bean
+    public PollableChannel s3Channel() {
+    	return new QueueChannel();
+    }
+}
+----
+
+> NOTE: Unlike the non-streaming inbound channel adapter, this adapter does not prevent duplicates by default.
+> If you do not delete the remote file and wish to prevent the file being processed again, you can configure an `S3PersistentFileListFilter` in the `filter` attribute.
+> If you don’t want to persist the state, an in-memory `SimpleMetadataStore` can be used with the filter.
+> If you wish to use a filename pattern (or regex) as well, use a `CompositeFileListFilter`.
+
+The `S3MessageHandler` is an Outbound Channel Adapter and allows performing `upload`, `download` and `copy` (see `S3MessageHandler.Command` enum) operations in the provided S3 bucket.
+
+The Java Configuration is:
+
+[source,java]
+----
+@SpringBootApplication
+public static class MyConfiguration {
+
+    @Autowired
+    private S3AsyncClient amazonS3;
+
+    @Bean
+    @ServiceActivator(inputChannel = "s3UploadChannel")
+    public MessageHandler s3MessageHandler() {
+    	return new S3MessageHandler(this.amazonS3, "my-bucket");
+    }
+
+}
+----
+
+With this config you can send a message with the `java.io.File` as `payload` and the `transferManager.upload()` operation will be performed, where the file name is used as a S3 Object key.
+
+See more information in the `S3MessageHandler` JavaDocs.
+
+NOTE: The AWS SDK recommends to use `S3CrtAsyncClient` for `S3TransferManager`, therefore an `S3AsyncClient.crtBuilder()` has to be used to achieve respective upload and download requirements, what is done internally in the `S3MessageHandler` when `S3CrtAsyncClient`-based constructor is used.
+
+The `S3MessageHandler` can be used as an Outbound Gateway with the `produceReply = true` constructor argument for Java Configuration.
+
+The "request-reply" nature of this gateway is async and the `Transfer` result from the `TransferManager` operation is sent to the `outputChannel`, assuming the transfer progress observation in the downstream flow.
+
+The `TransferListener` can be supplied to the `S3MessageHandler` to track the transfer progress per requests.
+
+See more information in the `S3MessageHandler` Javadocs.
+
+The Spring Integration dependency (as well as `s3-transfer-manager` and `aws-crt-client`) in the `spring-cloud-aws-s3` module are `optional` to avoid unnecessary artifacts on classpath when Spring Integration is not used.
+For convenience, a dedicated `spring-cloud-aws-starter-integration-s3` is provided managing all the required dependencies for Spring Integration support with Amazon S3.
+