Kinesis last checkpoint and shard iterator are Strings, but are parsed as BigIntegers and fail with NumberFormatException

[fran-jagic-infinum]

This issue is present in version 3.0.3, the error was not occurring in 2.5.2.

When processing a Kinesis record throws an exception,
the last checkpoint and shard iterator are both Strings but are assumed to be numeric. This causes an error when converting them to BigInteger, and masks the root exception which is never logged.

Stack trace:

Exception in thread "<name>-kinesis-consumer-76843" java.lang.NumberFormatException: For input string: "AAAAAA"
at java.base/java.lang.NumberFormatException.forInputString(Unknown Source)
at java.base/java.lang.Integer.parseInt(Unknown Source)
at java.base/java.math.BigInteger.<init>(Unknown Source)
at java.base/java.math.BigInteger.<init>(Unknown Source)
at org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter$ShardConsumer.rewindIteratorOnError(KinesisMessageDrivenChannelAdapter.java:1161)
at org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter$ShardConsumer.lambda$processTask$8(KinesisMessageDrivenChannelAdapter.java:1098)

and

Exception in thread "<name>-kinesis-consumer-90642" java.lang.NumberFormatException: Illegal embedded sign character
at java.base/java.math.BigInteger.<init>(Unknown Source)
at java.base/java.math.BigInteger.<init>(Unknown Source)
at org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter$ShardConsumer.rewindIteratorOnError(KinesisMessageDrivenChannelAdapter.java:1161)
at org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter$ShardConsumer.lambda$processTask$8(KinesisMessageDrivenChannelAdapter.java:1098)

Problem is in KinesisMessageDrivenChannelAdapter:1161

else if (lastCheckpoint == null
        || new BigInteger(lastCheckpoint).compareTo(new BigInteger(this.shardIterator)) < 0) {

[artembilan]

It looks like something is off with the nextShardIterator from the Kinesis response:

AAAAAAAAAAFld97aVDdSv5d5hCepMqW11oUiCUNFuxxYrc3JXkjPzj0moPPtQogaHfGHQp+71U4t7PmO2Oe8pnnRXg5D2RDQHvXRxAIss+RZXtkbDthUfUITKDiYZwMGOj7ilQTOapQKpW6fQqf+VtT1BEHTHl7yF18qoqvuXSFzwJ8QabarMbjO5URFLexS/SHJ/s4idPw=

It is really not a number.

I'll try to investigate what to do with that form now on.
It really was a number representation in AWS SDK v1...

[artembilan]

Just committed the fix.
Would be great if you can test it on your side against latest 3.0.5-SNAPSHOT.

Thanks

[fran-jagic-infinum]

Thank you very much, I'll test the fix today or tomorrow and post the results here.

[fran-jagic-infinum]

Hi, Artem,

Thank you very much for the quick response on this issue, and for maintaining and providing the library.

Unfortunately, testing this fix has proven difficult as the error is intermittent, and without the root cause I cannot reproduce it in our test environment. While I am working on that, could you please answer me a few things about the code in question?

1. In the fix code, I see there is still a BigInteger conversion for lastCheckpoint, is that value guaranteed to be numeric?
2. From what I can tell, rewind behaviour on record processing exception is new, and was introduced in v3.0. Before that, Kinesis records that failed to process were discarded? If this is a change in behaviour, it might be a problem for us. Is there a way to disable it?
3. What are the compatibility issues between Spring Boot 3 and spring-integration-aws 2.5.4? For Kinesis adapter would it be feasible for us to downgrade until we can reproduce and test this issue?

Thank you very much for your help!

[artembilan]

In the fix code, I see there is still a BigInteger conversion for lastCheckpoint, is that value guaranteed to be numeric?

Yes, the offset of the consumed record is always numeric. I took the logic from KCL to obtain the last offset in the consumed records batch.

Before that, Kinesis records that failed to process were discarded?

Well, that's exactly what was that fix for: do not lose records in case of error.

Is there a way to disable it?

No, we cannot disable that since our goal is to not lose records no matter the reason.

What are the compatibility issues between Spring Boot 3 and spring-integration-aws 2.5.4?

That's only possible to try on your own. There was a change in Spring Boot 3.0 how they resolve auto-configuration classes.
Looks like Spring Cloud AWS 2.4.x still uses spring.factories strategy.