From bc9a0825fbdb505c3f5690cd6725e027ba153072 Mon Sep 17 00:00:00 2001 From: abilan Date: Tue, 25 Apr 2023 17:11:07 -0400 Subject: [PATCH] GH-221: Revise an at-least-once delivery Fixes https://github.com/spring-projects/spring-integration-aws/issues/221 * Modify the logic of the `KinesisMessageDrivenChannelAdapter` to rewind a shard iterator to the failed sequence for any errors. A rewinding sequence is determined from extra properties in the `ShardCheckpointer` * Remove `RequestShardForSequenceException` since more natural behavior to react for any record processor error without end-user interaction --- README.md | 5 +- .../KinesisMessageDrivenChannelAdapter.java | 86 ++++++++----------- .../RequestShardForSequenceException.java | 48 ----------- .../inbound/kinesis/ShardCheckpointer.java | 35 ++++++-- .../aws/kinesis/KinesisIntegrationTests.java | 5 +- 5 files changed, 72 insertions(+), 107 deletions(-) delete mode 100644 src/main/java/org/springframework/integration/aws/inbound/kinesis/RequestShardForSequenceException.java diff --git a/README.md b/README.md index 1b5590e..74ac3b7 100644 --- a/README.md +++ b/README.md @@ -433,8 +433,9 @@ For example, users may want to fully read any parent shards before starting to r } ``` -Starting with _version 3.0_. the `RequestShardForSequenceException` can be used for flow control to request the shard iterator for specific sequence. -For example, when consumer has failed processing batch at specific record, throwing this exception with a sequence of that record will ensure at-least-once delivery since the shard iterator will move back to the requested record sequence. +Starting with _version 3.0_, any exception thrown from the record process may lead to shard iterator rewinding to the latest check-pointed sequence or the first one in the current failed batch. +This ensures an at-least-once delivery for possibly failed records. +If the latest checkpoint is equal to the highest sequence in the batch, then shard consumer continue with the next iterator. Also, the `KclMessageDrivenChannelAdapter` is provided for performing streams consumption by [Kinesis Client Library][]. See its JavaDocs for more information. 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 ae62137..9aba2d2 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,6 +65,7 @@ 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; @@ -1094,18 +1095,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport this.shardIterator = result.nextShardIterator(); } } - catch (RequestShardForSequenceException requestShardForSequenceException) { - // Something wrong happened and not all records were processed. - // Must start from the provided sequence - KinesisShardOffset newOffset = new KinesisShardOffset(this.shardOffset); - newOffset.setSequenceNumber(requestShardForSequenceException.getSequenceNumber()); - newOffset.setIteratorType(ShardIteratorType.AT_SEQUENCE_NUMBER); - GetShardIteratorRequest shardIteratorRequest = newOffset.toShardIteratorRequest(); - this.shardIterator = - KinesisMessageDrivenChannelAdapter.this.amazonKinesis - .getShardIterator(shardIteratorRequest) - .join() - .shardIterator(); + catch (Exception ex) { + rewindIteratorOnError(ex, result); } finally { attributesHolder.remove(); @@ -1154,6 +1145,39 @@ 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)) { + 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 { + newOffsetValue = this.checkpointer.getFirstSequenceInBatch(); + newOffset.setIteratorType(ShardIteratorType.AT_SEQUENCE_NUMBER); + } + + 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) { try { this.checkpointer.checkpoint(endingSequenceNumber); @@ -1205,6 +1229,7 @@ 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)) { @@ -1313,42 +1338,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport Message messageToSend = messageBuilder.build(); setAttributesIfNecessary(rawRecord, messageToSend); - try { - sendMessage(messageToSend); - } - catch (RequestShardForSequenceException requestShardForSequenceException) { - // Rethrow - throw requestShardForSequenceException; - } - catch (Exception ex) { - RequestShardForSequenceException requestShardForSequenceExceptionInCause = - findRequestShardForSequenceExceptionInCause(ex); - if (requestShardForSequenceExceptionInCause != null) { - throw requestShardForSequenceExceptionInCause; - } - else { - logger.info(ex, () -> - "Got an exception during sending a '" - + messageToSend - + "'" - + "\nfor the '" - + rawRecord - + "'.\n" - + "Consider to use 'errorChannel' flow for the compensation logic."); - } - } - } - - @Nullable - private static RequestShardForSequenceException findRequestShardForSequenceExceptionInCause(Throwable ex) { - if (ex instanceof RequestShardForSequenceException requestShardForSequenceException) { - return requestShardForSequenceException; - } - Throwable cause = ex.getCause(); - if (cause != null && cause != ex) { - return findRequestShardForSequenceExceptionInCause(cause); - } - return null; + sendMessage(messageToSend); } private void checkpointIfBatchMode() { diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/RequestShardForSequenceException.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/RequestShardForSequenceException.java deleted file mode 100644 index f815a2e..0000000 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/RequestShardForSequenceException.java +++ /dev/null @@ -1,48 +0,0 @@ -/* - * Copyright 2023 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.aws.inbound.kinesis; - -import software.amazon.awssdk.services.kinesis.model.ShardIteratorType; - -/** - * The flow control exception to notify the {@link KinesisMessageDrivenChannelAdapter} - * that specific shard iterator ({@link ShardIteratorType#AT_SEQUENCE_NUMBER}) - * must be requested instead of checkpointing. - * - * @author Artem Bilan - * - * @since 3.0 - */ -@SuppressWarnings("serial") -public class RequestShardForSequenceException extends RuntimeException { - - private final String sequenceNumber; - - public RequestShardForSequenceException(String sequenceNumber) { - this.sequenceNumber = sequenceNumber; - } - - public RequestShardForSequenceException(String sequenceNumber, Throwable cause) { - super(cause); - this.sequenceNumber = sequenceNumber; - } - - public String getSequenceNumber() { - return this.sequenceNumber; - } - -} 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 9b6c90e..eca01c5 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 @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2023 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -23,6 +23,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.integration.metadata.ConcurrentMetadataStore; import org.springframework.integration.metadata.MetadataStore; +import org.springframework.lang.Nullable; /** * An internal {@link Checkpointer} implementation based on provided {@link MetadataStore} @@ -42,6 +43,10 @@ class ShardCheckpointer implements Checkpointer { private final String key; + private volatile String firstSequenceInBatch; + + private volatile String highestSequence; + private volatile String lastCheckpointValue; private volatile boolean active = true; @@ -53,7 +58,7 @@ class ShardCheckpointer implements Checkpointer { @Override public boolean checkpoint() { - return checkpoint(this.lastCheckpointValue); + return checkpoint(this.highestSequence); } @Override @@ -66,7 +71,11 @@ class ShardCheckpointer implements Checkpointer { return this.checkpointStore.replace(this.key, existingSequence, sequenceNumber); } else { - return this.checkpointStore.putIfAbsent(this.key, sequenceNumber) == null; + boolean stored = this.checkpointStore.putIfAbsent(this.key, sequenceNumber) == null; + if (stored) { + this.lastCheckpointValue = sequenceNumber; + } + return stored; } } } @@ -79,14 +88,30 @@ class ShardCheckpointer implements Checkpointer { return false; } + void setFirstSequenceInBatch(String firstSequenceInBatch) { + this.firstSequenceInBatch = firstSequenceInBatch; + } + + @Nullable + String getFirstSequenceInBatch() { + return this.firstSequenceInBatch; + } + void setHighestSequence(String highestSequence) { - this.lastCheckpointValue = highestSequence; + this.highestSequence = highestSequence; } + String getHighestSequence() { + return this.highestSequence; + } + + @Nullable String getCheckpoint() { - return this.checkpointStore.get(this.key); + this.lastCheckpointValue = this.checkpointStore.get(this.key); + return this.lastCheckpointValue; } + @Nullable String getLastCheckpointValue() { return this.lastCheckpointValue; } diff --git a/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java b/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java index 4b24e90..0645deb 100644 --- a/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java +++ b/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java @@ -36,7 +36,6 @@ import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.aws.LocalstackContainerTest; import org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter; import org.springframework.integration.aws.inbound.kinesis.KinesisMessageHeaderErrorMessageStrategy; -import org.springframework.integration.aws.inbound.kinesis.RequestShardForSequenceException; import org.springframework.integration.aws.outbound.KinesisMessageHandler; import org.springframework.integration.aws.support.AwsHeaders; import org.springframework.integration.channel.QueueChannel; @@ -224,9 +223,7 @@ public class KinesisIntegrationTests implements LocalstackContainerTest { @Override public void postSend(Message message, MessageChannel channel, boolean sent) { if (message instanceof ErrorMessage errorMessage && this.thrown.compareAndSet(false, true)) { - throw new RequestShardForSequenceException( - errorMessage.getHeaders().get(AwsHeaders.RAW_RECORD, Record.class).sequenceNumber(), - errorMessage.getPayload()); + throw (RuntimeException) errorMessage.getPayload(); } }