Move Kinesis batch checkpoint before sending

Since the listener may take some time for records processing,
there is a possibility that checkpoint will be stored too late
after the process and thus we are able to get the same records
in different channel adapter for the same shard, even if they are
in the same consumer group and use shared `MetadataStore`

This solution is some compromise for the current state of things and
has to be reconsidered in the future in favor of proper rebalance and
shard leader election solution

As a workaround for the duplicate records an additional
`@IdempotentReceiver` approach can be used

* Upgrade to Gradle 4.2.1, Checkstyle 8.3, AssertJ 3.8.0
* Fix race condition in the `KinesisMessageDrivenChannelAdapterTests`
This commit is contained in:
Artem Bilan
2017-10-19 13:43:06 -04:00
parent 43754e19c6
commit 41b389e574
6 changed files with 15 additions and 9 deletions

View File

@@ -205,6 +205,8 @@ public class KinesisMessageDrivenChannelAdapterTests {
// When resharding happens the describeStream() is performed again
verify(this.amazonKinesisForResharding, atLeast(2)).describeStream(any(DescribeStreamRequest.class));
this.reshardingChannelAdapter.stop();
}
@Configuration