GH-221: Revise an at-least-once delivery

Fixes #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

Author: artembilan

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. 

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

@@ -65,6 +65,7 @@
 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 @@ private Runnable processTask() {
 						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 @@ private Runnable processTask() {
 			};
 		}
 
+		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 @@ private void prepareSleepState() {
 		private void processRecords(List<Record> 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 @@ private void performSend(
 
 			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() {

src/main/java/org/springframework/integration/aws/inbound/kinesis/RequestShardForSequenceException.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.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 @@ public boolean checkpoint(String sequenceNumber) {
 					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 @@ public boolean checkpoint(String sequenceNumber) {
 		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;
 	}

src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java

@@ -36,7 +36,6 @@
 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 PollableChannel errorChannel() {
 				@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();
 					}
 				}