GH-190: Swallow checkpoint provisioning exception

Resolves #190

Swallows `ProvisionedThroughputExceededException` while checkpointing
exhausted shards to avoid the `ShardConsumer` from not being marked as
closed and therefore be left in an inconsistent state which will only
throw exceptions as the `shardIterator` would be `null` and the
`ShardConsumer` wouldn't be marked as `CLOSED`.

Author: acm19

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2020 the original author or authors.
+ * Copyright 2017-2021 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.
@@ -93,6 +93,7 @@
  * @author Hervé Fortin
  * @author Dirk Bonhomme
  * @author Greg Eales
+ * @author Asiel Caballero
  *
  * @since 1.1
  */
@@ -1091,7 +1092,7 @@ private Runnable processTask() {
 										String endingSequenceNumber =
 												shard.getSequenceNumberRange().getEndingSequenceNumber();
 										if (endingSequenceNumber != null) {
-											this.checkpointer.checkpoint(endingSequenceNumber);
+											checkpointSwallowingProvisioningExceptions(endingSequenceNumber);
 										}
 										break;
 									}
@@ -1125,6 +1126,19 @@ private Runnable processTask() {
 			};
 		}
 
+		private void checkpointSwallowingProvisioningExceptions(String endingSequenceNumber) {
+			try {
+				this.checkpointer.checkpoint(endingSequenceNumber);
+			}
+			catch (ProvisionedThroughputExceededException ignored) {
+				// This exception is ignored to gurantee that an exhausted shard is marked as CLOSED
+				// even in the case it's not possible to checkpoint. Otherwise the ShardConsumer is
+				// left in an illegal state where the shard iterator is null without any possibility
+				// of recovering from it.
+				logger.debug("Exception while checkpointing empty shards", ignored);
+			}
+		}
+
 		private GetRecordsResult getRecords(GetRecordsRequest getRecordsRequest) {
 			try {
 				return KinesisMessageDrivenChannelAdapter.this.amazonKinesis.getRecords(getRecordsRequest);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2020 the original author or authors.
+ * Copyright 2017-2021 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.
@@ -75,6 +75,7 @@
  * @author Artem Bilan
  * @author Matthias Wesolowski
  * @author Greg Eales
+ * @author Asiel Caballero
  *
  * @since 1.1
  */
@@ -237,7 +238,6 @@ void testResharding() throws InterruptedException {
 	@Configuration
 	@EnableIntegration
 	public static class Config {
-
 		private final AtomicReference<KinesisShardEndedEvent> shardEndedEventReference = new AtomicReference<>();
 
 		@Bean
@@ -440,7 +440,7 @@ private void setNewShard(AmazonKinesis amazonKinesis, String shardIndex) {
 
 		@Bean
 		public ConcurrentMetadataStore reshardingCheckpointStore() {
-			return new SimpleMetadataStore();
+			return new ExceptionReadyMetadataStore();
 		}
 
 		@Bean
@@ -472,4 +472,14 @@ public void handleKinesisShardEndedEvent(KinesisShardEndedEvent event) {
 
 	}
 
+	private static class ExceptionReadyMetadataStore extends SimpleMetadataStore {
+		@Override
+		public boolean replace(String key, String oldValue, String newValue) {
+			if ("SpringIntegration:streamForResharding:closedShard4".equals(key)) {
+				throw new ProvisionedThroughputExceededException("Throughput exceeded");
+			}
+
+			return super.replace(key, oldValue, newValue);
+		}
+	}
 }