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

GH-238: Fix KinesisMDCA for highest offset

Browse Source 
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`
This commit is contained in:
Artem Bilan
2023-12-04 11:36:24 -05:00
parent 87a88f3097
commit 3e0352ef5a
1 changed files with 11 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java
View File

@@ -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<Record> 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);