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

GH-175: Iterate listShards() pages

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

`KinesisMessageDrivenChannelAdapter` doesn't read from all shards

* Add loop to get all pages of shards from Kinesis
* Upgrade to the latest SC-AWS
This commit is contained in:
Greg Eales
2020-10-19 11:47:57 +01:00
committed by Artem Bilan
parent ee79a1c632
commit c53bdfbf4c
2 changed files with 13 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -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'

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

@@ -42,8 +42,6 @@ import java.util.concurrent.TimeUnit;
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.locks.LockRegistry;
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 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
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) {