From 3e0352ef5a74f3a7245c00757a644ed70131867d Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 4 Dec 2023 11:36:24 -0500 Subject: [PATCH] GH-238: Fix KinesisMDCA for highest offset Fixes https://github.com/spring-projects/spring-integration-aws/issues/238 The `shardIterator` from Kinesis reply for records is not a sequence number anymore in AWS SKD v2. * Fix `rewindIteratorOnError()` logic to extract the highest offset from the current failed `GetRecordsResponse` to see if we have checkpointed it or we need to re-request the current `shardIterator` --- .../KinesisMessageDrivenChannelAdapter.java | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java index bd3a398..2f111ea 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java @@ -1158,9 +1158,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport "Ignore since the highest sequence in batch was check-pointed."); this.shardIterator = result.nextShardIterator(); } - else if (lastCheckpoint == null - || new BigInteger(lastCheckpoint).compareTo(new BigInteger(this.shardIterator)) < 0) { - + else if (reRequestCurrentShardIterator(lastCheckpoint, result)) { // No checkpoints for the shard - reuse the current shard iterator. logger.info(ex, "Record processor has thrown exception. " + "No checkpoints - re-request with the current shard iterator."); @@ -1183,6 +1181,16 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport } } + private boolean reRequestCurrentShardIterator(@Nullable String lastCheckpoint, GetRecordsResponse result) { + if (lastCheckpoint == null) { + return true; + } + List records = result.records(); + return !records.isEmpty() && + new BigInteger(lastCheckpoint) + .compareTo(new BigInteger(records.get(records.size() - 1).sequenceNumber())) < 0; + } + private void checkpointSwallowingProvisioningExceptions(String endingSequenceNumber) { try { this.checkpointer.checkpoint(endingSequenceNumber);