GH-175: Iterate listShards() pages

Fixes #175

`KinesisMessageDrivenChannelAdapter` doesn't read from all shards

* Add loop to get all pages of shards from Kinesis
* Upgrade to the latest SC-AWS

Author: Greg Eales

build.gradle

@@ -31,7 +31,7 @@ ext {
 	servletApiVersion = '4.0.1'
     localstackVersion = '0.1.22'
 	log4jVersion = '2.13.3'
-	springCloudAwsVersion = '2.2.3.RELEASE'
+	springCloudAwsVersion = '2.2.4.RELEASE'
 	springIntegrationVersion = '5.2.8.RELEASE'
 	kinesisClientVersion = '1.13.3'
 	kinesisProducerVersion = '0.14.1'

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

@@ -42,8 +42,6 @@
 import java.util.concurrent.locks.Lock;
 import java.util.stream.Collectors;
 
-import javax.annotation.Nullable;
-
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
@@ -64,6 +62,7 @@
 import org.springframework.integration.support.management.IntegrationManagedResource;
 import org.springframework.jmx.export.annotation.ManagedOperation;
 import org.springframework.jmx.export.annotation.ManagedResource;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.scheduling.SchedulingAwareRunnable;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
@@ -542,8 +541,17 @@ private List<Shard> readShardList(String stream, int retryCount) {
 
 		try {
 			ListShardsResult listShardsResult = this.amazonKinesis.listShards(listShardsRequest);
-
-			shardList.addAll(listShardsResult.getShards());
+			while (true) {
+				shardList.addAll(listShardsResult.getShards());
+				if (listShardsResult.getNextToken() == null) {
+					break;
+				}
+				else {
+					listShardsResult =
+							this.amazonKinesis.listShards(new ListShardsRequest()
+									.withNextToken(listShardsResult.getNextToken()));
+				}
+			}
 
 		}
 		catch (LimitExceededException limitExceededException) {