From d7a2dd590d3e9b9799c58b34e679b83d694191ab Mon Sep 17 00:00:00 2001 From: abilan Date: Mon, 8 May 2023 12:19:50 -0400 Subject: [PATCH] Fix KinesisMDChA rewindIteratorOnError() for NPE The error in the `KinesisMessageDrivenChannelAdapter.ShardConsumer.processTask()` might be also thrown just direct from the `amazonKinesis.getRecords(getRecordsRequest)`. If it happens first time, the `ShardCheckpointer` is not initialized with sequence numbers. Therefore, a condition `highestSequence.equals(lastCheckpoint)` may lead to NPE. * Rework the logic in the `KinesisMessageDrivenChannelAdapter.ShardConsumer.rewindIteratorOnError()` to deal with a `null` for `this.checkpointer.getHighestSequence()` and reuse the current `shardIterator` in the next request if no any commits happened. * Remove `ShardCheckpointer.firstSequenceInBatch` since this is exactly a meaning of `shardIterator` representation. Related to: https://github.com/spring-projects/spring-integration-aws/issues/223 --- .../KinesisMessageDrivenChannelAdapter.java | 50 ++++++++++--------- .../inbound/kinesis/ShardCheckpointer.java | 12 +---- 2 files changed, 28 insertions(+), 34 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 63ef44f..bd3a398 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 @@ -65,7 +65,6 @@ import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.core.AttributeAccessor; import org.springframework.core.convert.converter.Converter; -import org.springframework.core.log.LogMessage; import org.springframework.core.serializer.support.DeserializingConverter; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.aws.event.KinesisShardEndedEvent; @@ -1146,36 +1145,42 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport } private void rewindIteratorOnError(Exception ex, GetRecordsResponse result) { - KinesisShardOffset newOffset = new KinesisShardOffset(this.shardOffset); String lastCheckpoint = this.checkpointer.getLastCheckpointValue(); String highestSequence = this.checkpointer.getHighestSequence(); - if (highestSequence.equals(lastCheckpoint)) { + + if (highestSequence == null) { + // Haven't reached record process - reuse the current shard iterator. + logger.info(ex, "getRecords request has thrown exception. " + + "No checkpoints - re-request with the current shard iterator."); + } + else if (highestSequence.equals(lastCheckpoint)) { logger.info(ex, "Record processor has thrown exception. " + "Ignore since the highest sequence in batch was check-pointed."); this.shardIterator = result.nextShardIterator(); - return; } - String newOffsetValue = lastCheckpoint; - if (lastCheckpoint != null) { - newOffset.setIteratorType(ShardIteratorType.AFTER_SEQUENCE_NUMBER); + else if (lastCheckpoint == null + || new BigInteger(lastCheckpoint).compareTo(new BigInteger(this.shardIterator)) < 0) { + + // 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."); } else { - newOffsetValue = this.checkpointer.getFirstSequenceInBatch(); - newOffset.setIteratorType(ShardIteratorType.AT_SEQUENCE_NUMBER); + KinesisShardOffset newOffset = new KinesisShardOffset(this.shardOffset); + newOffset.setIteratorType(ShardIteratorType.AFTER_SEQUENCE_NUMBER); + + logger.info(ex, () -> + "Record processor has thrown exception. " + + "Rewind shard iterator after sequence number: " + lastCheckpoint); + + newOffset.setSequenceNumber(lastCheckpoint); + GetShardIteratorRequest shardIteratorRequest = newOffset.toShardIteratorRequest(); + this.shardIterator = + KinesisMessageDrivenChannelAdapter.this.amazonKinesis + .getShardIterator(shardIteratorRequest) + .join() + .shardIterator(); } - - logger.info(ex, - LogMessage.format("Record processor has thrown exception. " + - "Rewind shard iterator %s sequence number: %s", - (lastCheckpoint != null ? "after" : "at"), newOffsetValue)); - - newOffset.setSequenceNumber(newOffsetValue); - GetShardIteratorRequest shardIteratorRequest = newOffset.toShardIteratorRequest(); - this.shardIterator = - KinesisMessageDrivenChannelAdapter.this.amazonKinesis - .getShardIterator(shardIteratorRequest) - .join() - .shardIterator(); } private void checkpointSwallowingProvisioningExceptions(String endingSequenceNumber) { @@ -1235,7 +1240,6 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport private void processRecords(List records) { logger.trace(() -> "Processing records: " + records + " for [" + ShardConsumer.this + "]"); - this.checkpointer.setFirstSequenceInBatch(records.get(0).sequenceNumber()); this.checkpointer.setHighestSequence(records.get(records.size() - 1).sequenceNumber()); if (ListenerMode.record.equals(KinesisMessageDrivenChannelAdapter.this.listenerMode)) { diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java index eca01c5..ce9681d 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java @@ -43,8 +43,6 @@ class ShardCheckpointer implements Checkpointer { private final String key; - private volatile String firstSequenceInBatch; - private volatile String highestSequence; private volatile String lastCheckpointValue; @@ -88,19 +86,11 @@ class ShardCheckpointer implements Checkpointer { return false; } - void setFirstSequenceInBatch(String firstSequenceInBatch) { - this.firstSequenceInBatch = firstSequenceInBatch; - } - - @Nullable - String getFirstSequenceInBatch() { - return this.firstSequenceInBatch; - } - void setHighestSequence(String highestSequence) { this.highestSequence = highestSequence; } + @Nullable String getHighestSequence() { return this.highestSequence; }