Spring/spring-integration-aws
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
Fixes https://github.com/spring-projects/spring-integration-aws/issues/179

* code review
* rename to `shardListFilter` and add example to README
This commit is contained in:
Greg Eales
2020-10-20 17:59:44 +01:00
committed by GitHub
parent c53bdfbf4c
commit 5a3c6a1a95
2 changed files with 36 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -40,6 +40,7 @@ import java.util.concurrent.Future;
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 class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
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 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
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 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
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) {