GH-179: Kinesis: Add "shards to consume" filter

Fixes #179

* code review
* rename to `shardListFilter` and add example to README

Author: 0x006EA1E5

README.md

@@ -580,7 +580,22 @@ The `KinesisMessageDrivenChannelAdapter` iterates over its shards and tries to a
 If `LockRegistry` is not provided, no exclusive locking happens and all the shards are consumed by this `KinesisMessageDrivenChannelAdapter`. 
 See also `DynamoDbLockRegistry` for more information.
 
-Also the `KclMessageDrivenChannelAdapter` is provided for performing streams consumption by [Kinesis Client Library][].
+The `KinesisMessageDrivenChannelAdapter` can be configured with a `Function<List<Shard>, List<Shard>> shardListFilter` to filter the available, open, non-exhausted shards.
+This filter `Function` will be called each time the shard list is refreshed.
+
+For example, users may want to fully read any parent shards before starting to read their child shards.  This could be achieved as follows:
+
+```java
+    openShards -> {
+        Set<String> openShardIds = openShards.stream().map(Shard::getShardId).collect(Collectors.toSet());
+        // only return open shards which have no parent available for reading 
+        return openShards.stream()
+                .filter(shard -> !openShardIds.contains(shard.getParentShardId()))
+                .collect(Collectors.toList());
+        }
+```
+
+Also, the `KclMessageDrivenChannelAdapter` is provided for performing streams consumption by [Kinesis Client Library][].
 See its JavaDocs for more information. 
 
 ### Outbound Channel Adapter

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

@@ -40,6 +40,7 @@
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import org.springframework.beans.factory.DisposableBean;
@@ -172,6 +173,9 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
 
 	private ApplicationEventPublisher applicationEventPublisher;
 
+	@Nullable
+	private Function<List<Shard>, List<Shard>> shardListFilter;
+
 	public KinesisMessageDrivenChannelAdapter(AmazonKinesis amazonKinesis, String... streams) {
 		Assert.notNull(amazonKinesis, "'amazonKinesis' must not be null.");
 		Assert.notEmpty(streams, "'streams' must not be null.");
@@ -339,6 +343,15 @@ public void setBindSourceRecord(boolean bindSourceRecord) {
 		this.bindSourceRecord = bindSourceRecord;
 	}
 
+	/**
+	 * Specify a {@link Function Function&lt;List&lt;Shard&gt;, List&lt;Shard&gt;&gt;} to filter the shards which will
+	 * be read from.
+	 * @param shardListFilter the filter {@link Function Function&lt;List&lt;Shard&gt;, List&lt;Shard&gt;&gt;}
+	 */
+	public void setShardListFilter(Function<List<Shard>, List<Shard>> shardListFilter) {
+		this.shardListFilter = shardListFilter;
+	}
+
 	@Override
 	protected void onInit() {
 		super.onInit();
@@ -618,19 +631,19 @@ private List<Shard> detectShardsToConsume(String stream, int retry) {
 				if (endingSequenceNumber != null) {
 					String checkpoint = this.checkpointStore.get(key);
 
-					boolean skipClosedShard = checkpoint != null && new BigInteger(endingSequenceNumber)
+					boolean skipClosedAndExhaustedShard = checkpoint != null && new BigInteger(endingSequenceNumber)
 							.compareTo(new BigInteger(checkpoint)) <= 0;
 
 					if (logger.isTraceEnabled()) {
 						logger.trace("The shard [" + shard + "] in stream [" + stream
-								+ "] is closed CLOSED with endingSequenceNumber [" + endingSequenceNumber
+								+ "] is closed CLOSED and exhausted with endingSequenceNumber [" + endingSequenceNumber
 								+ "].\nThe last processed checkpoint is [" + checkpoint + "]."
-								+ (skipClosedShard ? "\nThe shard will be skipped." : ""));
+								+ (skipClosedAndExhaustedShard ? "\nThe shard will be skipped." : ""));
 					}
 
-					if (skipClosedShard) {
-						// Skip CLOSED shard which has been read before
-						// according a checkpoint
+					if (skipClosedAndExhaustedShard) {
+						// Skip CLOSED shard which has been exhausted
+						// according the checkpoint
 						continue;
 					}
 				}
@@ -649,8 +662,7 @@ private List<Shard> detectShardsToConsume(String stream, int retry) {
 			sleep(this.describeStreamBackoff, new IllegalStateException(exceptionMessage), false);
 		}
 
-		return shardsToConsume;
-
+		return this.shardListFilter != null ? this.shardListFilter.apply(shardsToConsume) : shardsToConsume;
 	}
 
 	private void sleep(long sleepAmount, RuntimeException error, boolean interruptThread) {