Spring/spring-integration-aws
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-221: Revise the manual checkpoint logic

Browse Source 
Fixes https://github.com/spring-projects/spring-integration-aws/issues/221

* Introduce a `RequestShardForSequenceException` to control the flow for
requesting the `KinesisMessageDrivenChannelAdapter` for shard iterator at specific sequence
This commit is contained in:
abilan
2023-04-24 17:20:23 -04:00
parent 136cce6177
commit 63b40836ce
7 changed files with 133 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
README.md
View File

@@ -398,7 +398,7 @@ Can be specified as `null` with meaning no conversion and the target `Message` i
Additional headers like `AwsHeaders.RECEIVED_STREAM`, `AwsHeaders.SHARD`, `AwsHeaders.RECEIVED_PARTITION_KEY` and `AwsHeaders.RECEIVED_SEQUENCE_NUMBER` are populated to the message for downstream logic.
When `CheckpointMode.manual` is used the `Checkpointer` instance is populated to the `AwsHeaders.CHECKPOINTER` header for an acknowledgment in the downstream logic manually.
The `KinesisMessageDrivenChannelAdapter` ca be configured with the `ListenerMode` `record` or `batch` to process records one by one or send the whole just polled batch of records.
The `KinesisMessageDrivenChannelAdapter` can be configured with the `ListenerMode` `record` or `batch` to process records one by one or send the whole just polled batch of records.
If `Converter` is configured to `null`, the entire `List<Record>` is sent as a payload.
Otherwise, a list of converted `Record.getData().array()` is wrapped to the payload of message to send.
In this case the `AwsHeaders.RECEIVED_PARTITION_KEY` and `AwsHeaders.RECEIVED_SEQUENCE_NUMBER` headers contains values as a `List<String>` of partition keys and sequence numbers of converted records respectively.
@@ -433,6 +433,9 @@ For example, users may want to fully read any parent shards before starting to r
}
```
Starting with _version 3.0_. the `RequestShardForSequenceException` can be used for flow control to request the shard iterator for specific sequence.
For example, when consumer has failed processing batch at specific record, throwing this exception with a sequence of that record will ensure at-least-once delivery since the shard iterator will move back to the requested record sequence.
Also, the `KclMessageDrivenChannelAdapter` is provided for performing streams consumption by [Kinesis Client Library][].
See its JavaDocs for more information.