<commit_before>[[spring-cloud-stream-binder-kinesis-reference]]
= Spring Cloud Stream AWS Kinesis Binder Reference Guide

--
This guide describes the https://aws.amazon.com/kinesis/[AWS Kinesis] implementation of the Spring Cloud Stream Binder.
It contains information about its design, usage and configuration options, as well as information on how the Stream Cloud Stream concepts map into AWS Kinesis specific constructs.
--

== Usage

For using the AWS Kinesis Binder, you just need to add it to your Spring Cloud Stream application, using the following Maven coordinates:

[source,xml]
----
<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-stream-binder-kinesis</artifactId>
</dependency>
----

== 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:

.Kinesis Binder
image::{github-raw}/docs/src/main/asciidoc/images/kinesis-binder.png[width=300,scaledwidth="50%"]

Unlike https://kafka.apache.org/[Apache Kafka] the AWS Kinesis doesn't provide out-of-the-box support for consumer groups.
The support of this feature is implemented as a part of `MetadataStore` key for shard checkpoints in the `KinesisMessageDrivenChannelAdapter` - `[CONSUMER_GROUP]:[STREAM]:[SHARD_ID]`.
In addition, the `LockRegistry` is used to ensure exclusive access to each shard.
This way only one channel adapter in the same consumer group will consumer messages from a single shard in the stream it is configured for.

The partitioning logic in AWS Kinesis is similar to the Apache Kafka support, but with slightly different logic.
The `partitionKey` on the producer side determines which shard in the stream the data record is assigned to.
Partition keys are Unicode strings with a maximum length limit of 256 characters for each key.
AWS Kinesis uses the partition key as input to a hash function that maps the partition key and associated data to a specific shard.
Specifically, an MD5 hash function is used to map partition keys to 128-bit integer values and to map associated data records to shards.
As a result of this hashing mechanism, all data records with the same partition key map to the same shard within the stream.
But at the same time we can't select target shard to send explicitly.
Although calculating the hash manually (and use `explicitHashKeyExpression` for producer, respectively), we may track the target shard by inclusion into its `HashKeyRange`.

By default partition key is a result of the `Object.hash()` from the message `payload`.

The Spring Cloud Stream partition handling logic is excluded in case of AWS Kinesis Binder since it is out of use and the provided `producer.partitionKeyExpression` is propagated to the `KinesisMessageHandler` directly.

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 only subscribe 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.
A `single consumer` for the message is ensured by https://docs.spring.io/spring-cloud-stream/docs/Elmhurst.RELEASE/reference/htmlsingle/#consumer-groups[consumer group abstraction].

To have a highly available consumer group for your kinesis stream:

- Ensure all instances of your consumer applications use a shared `DynamoDbMetadataStore` and `DynamoDbLockRegistry` (See below for configuration options).
- Use same group name for the channel in all application instances by using property `spring.cloud.stream.bindings.<bindingTarget>.group`.

These configurations alone guarantee HA, message ordering and guaranteed message delivery.
However, even distribution across instances is not guaranteed as of now.
There is a very high chance that a single instance in a consumer group will pick up all the shards for consuming.
But, when that instance goes down (couldn't send heartbeat for any reason), other instance in the consumer group will start processing from the last checkpoint of the previous consumer (for shardIterator type TRIM_HORIZON).

So, configuring consumer concurrency is important to achieve throughput.
It can be configured using `spring.cloud.stream.bindings.<bindingTarget>.consumer.concurrency`.

=== Static shard distribution within a single consumer group
It is possible to evenly distribute shard across all instances within a single consumer group.
This done by configuring:

- `spring.cloud.stream.instanceCount=` to number of instances
- `spring.cloud.stream.instanceIndex=` current instance's index

The only way to achieve HA in this case is that, when an instance processing a particular shard goes down, another instance must have `spring.cloud.stream.instanceIndex=` to be the same as the failed instance's index to start processing from those shards.

== Configuration Options

This section contains settings specific to the Kinesis Binder and bound channels.

For general binding configuration options and properties, please refer to the https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#_configuration_options[Spring Cloud Stream core documentation].

[[kinesis-binder-properties]]
=== Kinesis Binder Properties

The following properties are available for Kinesis Binder configuration, which start with the `spring.cloud.stream.kinesis.binder.` prefix

headers::
The set of custom headers to transfer over AWS Kinesis
+
Default: "correlationId", "sequenceSize", "sequenceNumber", "contentType", "originalContentType".
describeStreamBackoff::
The amount of time in milliseconds in between retries for the `DescribeStream` operation
+
Default: `1000`.
describeStreamRetries::
The amount of times the consumer will retry a `DescribeStream` operation waiting for the stream to be in `ACTIVE` state
+
Default: `50`.
autoCreateStream::
If set to `true`, the binder will create the stream automatically.
If set to `false`, the binder will rely on the stream being already created.
+
Default: `true`
autoAddShards::
If set to `true`, the binder will create new shards automatically.
If set to `false`, the binder will rely on the shard size of the stream being already configured.
If the shard count of the target stream is smaller than the expected value, the binder will ignore that value
+
Default: `false`
minShardCount::
Effective only if `autoAddShards` is set to `true`.
The minimum number of shards that the binder will configure on the stream from which it produces/consumes data.
It can be superseded by the `partitionCount` setting of the producer or by the value of `instanceCount * concurrency` settings of the producer (if either is larger)
+
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
+
Default: `false`
<commit_msg>Made a correction in kinesis doc<commit_after>[[spring-cloud-stream-binder-kinesis-reference]]
= Spring Cloud Stream AWS Kinesis Binder Reference Guide

--
This guide describes the https://aws.amazon.com/kinesis/[AWS Kinesis] implementation of the Spring Cloud Stream Binder.
It contains information about its design, usage and configuration options, as well as information on how the Stream Cloud Stream concepts map into AWS Kinesis specific constructs.
--

== Usage

For using the AWS Kinesis Binder, you just need to add it to your Spring Cloud Stream application, using the following Maven coordinates:

[source,xml]
----
<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-stream-binder-kinesis</artifactId>
</dependency>
----

== 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:

.Kinesis Binder
image::{github-raw}/docs/src/main/asciidoc/images/kinesis-binder.png[width=300,scaledwidth="50%"]

Unlike https://kafka.apache.org/[Apache Kafka] the AWS Kinesis doesn't provide out-of-the-box support for consumer groups.
The support of this feature is implemented as a part of `MetadataStore` key for shard checkpoints in the `KinesisMessageDrivenChannelAdapter` - `[CONSUMER_GROUP]:[STREAM]:[SHARD_ID]`.
In addition, the `LockRegistry` is used to ensure exclusive access to each shard.
This way only one channel adapter in the same consumer group will consumer messages from a single shard in the stream it is configured for.

The partitioning logic in AWS Kinesis is similar to the Apache Kafka support, but with slightly different logic.
The `partitionKey` on the producer side determines which shard in the stream the data record is assigned to.
Partition keys are Unicode strings with a maximum length limit of 256 characters for each key.
AWS Kinesis uses the partition key as input to a hash function that maps the partition key and associated data to a specific shard.
Specifically, an MD5 hash function is used to map partition keys to 128-bit integer values and to map associated data records to shards.
As a result of this hashing mechanism, all data records with the same partition key map to the same shard within the stream.
But at the same time we can't select target shard to send explicitly.
Although calculating the hash manually (and use `explicitHashKeyExpression` for producer, respectively), we may track the target shard by inclusion into its `HashKeyRange`.

By default partition key is a result of the `Object.hash()` from the message `payload`.

The Spring Cloud Stream partition handling logic is excluded in case of AWS Kinesis Binder since it is out of use and the provided `producer.partitionKeyExpression` is propagated to the `KinesisMessageHandler` directly.

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 only subscribe 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.
A `single consumer` for the message is ensured by https://docs.spring.io/spring-cloud-stream/docs/Elmhurst.RELEASE/reference/htmlsingle/#consumer-groups[consumer group abstraction].

To have a highly available consumer group for your kinesis stream:

- Ensure all instances of your consumer applications use a shared `DynamoDbMetadataStore` and `DynamoDbLockRegistry` (See below for configuration options).
- Use same group name for the channel in all application instances by using property `spring.cloud.stream.bindings.<bindingTarget>.group`.

These configurations alone guarantee HA, message ordering and guaranteed message delivery.
However, even distribution across instances is not guaranteed as of now.
There is a very high chance that a single instance in a consumer group will pick up all the shards for consuming.
But, when that instance goes down (couldn't send heartbeat for any reason), other instance in the consumer group will start processing from the last checkpoint of the previous consumer (for shardIterator type TRIM_HORIZON).

So, configuring consumer concurrency is important to achieve throughput.
It can be configured using `spring.cloud.stream.bindings.<bindingTarget>.consumer.concurrency`.

=== Static shard distribution within a single consumer group
It is possible to evenly distribute shard across all instances within a single consumer group.
This done by configuring:

- `spring.cloud.stream.instanceCount=` to number of instances
- `spring.cloud.stream.instanceIndex=` current instance's index

The only way to achieve HA in this case is that, when an instance processing a particular shard goes down, another instance must have `spring.cloud.stream.instanceIndex=` to be the same as the failed instance's index to start processing from those shards.

== Configuration Options

This section contains settings specific to the Kinesis Binder and bound channels.

For general binding configuration options and properties, please refer to the https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#_configuration_options[Spring Cloud Stream core documentation].

[[kinesis-binder-properties]]
=== Kinesis Binder Properties

The following properties are available for Kinesis Binder configuration, which start with the `spring.cloud.stream.kinesis.binder.` prefix

headers::
The set of custom headers to transfer over AWS Kinesis
+
Default: "correlationId", "sequenceSize", "sequenceNumber", "contentType", "originalContentType".
describeStreamBackoff::
The amount of time in milliseconds in between retries for the `DescribeStream` operation
+
Default: `1000`.
describeStreamRetries::
The amount of times the consumer will retry a `DescribeStream` operation waiting for the stream to be in `ACTIVE` state
+
Default: `50`.
autoCreateStream::
If set to `true`, the binder will create the stream automatically.
If set to `false`, the binder will rely on the stream being already created.
+
Default: `true`
autoAddShards::
If set to `true`, the binder will create new shards automatically.
If set to `false`, the binder will rely on the shard size of the stream being already configured.
If the shard count of the target stream is smaller than the expected value, the binder will ignore that value
+
Default: `false`
minShardCount::
Effective only if `autoAddShards` is set to `true`.
The minimum number of shards that the binder will configure on the stream from which it produces/consumes data.
It can be superseded by the `partitionCount` setting of the producer or by the value of `instanceCount * concurrency` settings of the consumer (if either is larger)
+
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
+
Default: `false`

{'fraction_non_alphanumeric': 0.05963332680889933, 'fraction_numerical': 0.0024792849220330137, 'mean_word_length': 5.235964198535394, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 19, 'https://': 14, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7151818', 'n_tokens_mistral': 3863, 'n_tokens_neox': 3723, 'n_words': 1953}