luciano/amazon-kinesis-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fixing bug where ShardFilter parameter for ListShards was being passe… (#788)

Browse source 
* Fixing bug where ShardFilter parameter for ListShards was being passed in for paginated calls.

This resulted in a bug where all calls for ListShards when initializing
the lease table would fail, since Kinesis only requires the NextToken
parameter when making paginated calls.

* Adding some logging for listShards.

Co-authored-by: Joshua Kim <kimjos@amazon.com>
This commit is contained in:
Joshua Kim 2021-02-18 18:26:24 -08:00 committed by GitHub
parent d5cac07851
commit 5d68e65c2c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 8 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/KinesisShardDetector.java
View file

@ -192,12 +192,16 @@ public class KinesisShardDetector implements ShardDetector {
exceptionManager.add(ResourceInUseException.class, t -> t); exceptionManager.add(ResourceInUseException.class, t -> t);
exceptionManager.add(KinesisException.class, t -> t); exceptionManager.add(KinesisException.class, t -> t);
ListShardsRequest.Builder request = KinesisRequestsBuilder.listShardsRequestBuilder().shardFilter(shardFilter); ListShardsRequest.Builder builder = KinesisRequestsBuilder.listShardsRequestBuilder();
if (StringUtils.isEmpty(nextToken)) { if (StringUtils.isEmpty(nextToken)) {
request = request.streamName(streamIdentifier.streamName()); builder = builder.streamName(streamIdentifier.streamName()).shardFilter(shardFilter);
} else { } else {
request = request.nextToken(nextToken); builder = builder.nextToken(nextToken);
} }
final ListShardsRequest request = builder.build();
log.info("Stream {}: listing shards with list shards request {}", streamIdentifier, request);
ListShardsResponse result = null; ListShardsResponse result = null;
LimitExceededException lastException = null; LimitExceededException lastException = null;
int remainingRetries = maxListShardsRetryAttempts; int remainingRetries = maxListShardsRetryAttempts;
@ -205,7 +209,7 @@ public class KinesisShardDetector implements ShardDetector {
while (result == null) { while (result == null) {
try { try {
try { try {
result = getListShardsResponse(request.build()); result = getListShardsResponse(request);
} catch (ExecutionException e) { } catch (ExecutionException e) {
throw exceptionManager.apply(e.getCause()); throw exceptionManager.apply(e.getCause());
} catch (InterruptedException e) { } catch (InterruptedException e) {