No NextShardIterator if not processed manual ack batch

* fix(KinesisMessageDrivenChannelAdapter): added logic to not use the NextShardIterator if Manual Checkpointer hasnt reached the last checkpoint

* fix(KinesisMessageDrivenChannel): added logs for getNextShardIterator

* fix(KinesisMessageDrivenChannelAdapter): fixed logic and comments issue

* feat(KinesisMessageDrivenChannelAdapter): implemented integration tests for getNextShardIterator new usecase

* feat(KinesisMessageDrivenChannelAdapter): updated @author

* feat(KinesisMessageDrivenChannelAdapterTests): added author data

Author: sonikro

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

@@ -94,6 +94,7 @@
  * @author Dirk Bonhomme
  * @author Greg Eales
  * @author Asiel Caballero
+ * @author Jonathan Nagayoshi
  *
  * @since 1.1
  */
@@ -1073,7 +1074,40 @@ private Runnable processTask() {
 				finally {
 					attributesHolder.remove();
 					if (result != null) {
-						this.shardIterator = result.getNextShardIterator();
+                         // If using manual checkpointer, we have to make sure we are allowed to use the next shard iterator
+                         // Because if the manual checkpointer was not set to the latest record, it means there are records to be reprocessed
+                         // and if we use the nextShardIterator, we will be skipping records that need to be reprocessed
+						List<Record> records = result.getRecords();
+						if (CheckpointMode.manual.equals(KinesisMessageDrivenChannelAdapter.this.checkpointMode) &&
+								!records.isEmpty()) {
+							logger.info("Manual checkpointer. Must validate if should use getNextShardIterator()");
+							String lastRecordSequence = records.get(records.size() - 1).getSequenceNumber();
+							String lastCheckpointSequence = this.checkpointer.getCheckpoint();
+							if (lastCheckpointSequence.equals(lastRecordSequence)) {
+								logger.info("latestCheckpointSequence is same as latestRecordSequence. " +
+										"" +
+										"Should getNextShardIterator()");
+								// Means the manual checkpointer has processed the last record, Should move forward
+								this.shardIterator = result.getNextShardIterator();
+							}
+							else {
+								logger.info("latestCheckpointSequence is not the same as latestRecordSequence" +
+										". Should Get a new iterator AFTER_SEQUENCE_NUMBER latestCheckpointSequence");
+								// Something wrong happened and not all records were processed.
+								// Must start from the latest known checkpoint
+								KinesisShardOffset newOffset = new KinesisShardOffset(this.shardOffset);
+								newOffset.setSequenceNumber(lastCheckpointSequence);
+								newOffset.setIteratorType(ShardIteratorType.AFTER_SEQUENCE_NUMBER);
+								GetShardIteratorRequest shardIteratorRequest = newOffset.toShardIteratorRequest();
+								this.shardIterator = KinesisMessageDrivenChannelAdapter.this
+										.amazonKinesis
+										.getShardIterator(shardIteratorRequest)
+										.getShardIterator();
+							}
+						}
+						else {
+							this.shardIterator = result.getNextShardIterator();
+						}
 
 						if (this.shardIterator == null) {
 							if (KinesisMessageDrivenChannelAdapter.this.lockRegistry != null) {

src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java

@@ -76,6 +76,7 @@
  * @author Matthias Wesolowski
  * @author Greg Eales
  * @author Asiel Caballero
+ * @author Jonathan Nagayoshi
  *
  * @since 1.1
  */
@@ -201,6 +202,40 @@ void testKinesisMessageDrivenChannelAdapter() {
 				.hasSize(2);
 
 		this.kinesisMessageDrivenChannelAdapter.stop();
+
+		this.kinesisMessageDrivenChannelAdapter.setListenerMode(ListenerMode.batch);
+		this.kinesisMessageDrivenChannelAdapter.setCheckpointMode(CheckpointMode.manual);
+		this.checkpointStore.put("SpringIntegration" + ":" + STREAM1 + ":" + "1", "2");
+
+		this.kinesisMessageDrivenChannelAdapter.start();
+
+		message = this.kinesisChannel.receive(10000);
+		assertThat(message).isNotNull();
+		assertThat(message.getPayload()).isInstanceOf(List.class);
+		List<String> messagePayload = (List<String>) message.getPayload();
+		assertThat(messagePayload).size().isEqualTo(3);
+
+		Object messageSequenceNumberHeader = message.getHeaders().get(AwsHeaders.RECEIVED_SEQUENCE_NUMBER);
+		assertThat(messageSequenceNumberHeader).isInstanceOf(List.class);
+		assertThat((List<String>) messageSequenceNumberHeader).contains("3");
+		// Set checkpoint to 3, this should prevent adapter from using next shard, since its not the latest record
+		// in the batch
+		checkpointer.checkpoint("3");
+
+		await().untilAsserted(
+				() -> assertThat(this.checkpointStore.get("SpringIntegration" + ":" + STREAM1 + ":" + "1"))
+						.isEqualTo("3"));
+		message = this.kinesisChannel.receive(10000);
+		assertThat(message).isNotNull();
+		assertThat(message.getPayload()).isInstanceOf(List.class);
+		messagePayload = (List<String>) message.getPayload();
+		assertThat(messagePayload).size().isEqualTo(2);
+		assertThat(messagePayload).contains("bar");
+		assertThat(messagePayload).contains("foobar");
+
+		this.kinesisMessageDrivenChannelAdapter.stop();
+
+
 	}
 
 	@Test
@@ -296,6 +331,36 @@ public AmazonKinesis amazonKinesis() {
 							.withRecords(new Record().withPartitionKey("partition1").withSequenceNumber("2")
 									.withData(ByteBuffer.wrap(serializingConverter.convert("bar")))));
 
+
+			String shard1Iterator5 = "shard1Iterator5";
+			String shard1Iterator6 = "shard1Iterator6";
+
+			given(amazonKinesis.getShardIterator(
+					KinesisShardOffset.afterSequenceNumber(STREAM1, "1", "2").toShardIteratorRequest()))
+					.willReturn(new GetShardIteratorResult().withShardIterator(shard1Iterator5));
+
+			given(amazonKinesis.getRecords(new GetRecordsRequest().withShardIterator(shard1Iterator5).withLimit(25)))
+					.willReturn(new GetRecordsResult().withNextShardIterator(shard1Iterator6)
+							.withRecords(new Record().withPartitionKey("partition1").withSequenceNumber("3")
+									.withData(ByteBuffer.wrap(serializingConverter.convert("foo"))),
+									new Record().withPartitionKey("partition1").withSequenceNumber("4")
+											.withData(ByteBuffer.wrap(serializingConverter.convert("bar"))),
+									new Record().withPartitionKey("partition1").withSequenceNumber("5")
+											.withData(ByteBuffer.wrap(serializingConverter.convert("foobar")))));
+
+
+			given(amazonKinesis.getShardIterator(
+					KinesisShardOffset.afterSequenceNumber(STREAM1, "1", "3").toShardIteratorRequest()))
+					.willReturn(new GetShardIteratorResult().withShardIterator(shard1Iterator6));
+
+			given(amazonKinesis.getRecords(new GetRecordsRequest().withShardIterator(shard1Iterator6).withLimit(25)))
+					.willReturn(new GetRecordsResult().withNextShardIterator(shard1Iterator6)
+							.withRecords(
+									new Record().withPartitionKey("partition1").withSequenceNumber("4")
+											.withData(ByteBuffer.wrap(serializingConverter.convert("bar"))),
+									new Record().withPartitionKey("partition1").withSequenceNumber("5")
+											.withData(ByteBuffer.wrap(serializingConverter.convert("foobar")))));
+
 			return amazonKinesis;
 		}