From 5d68e65c2c7c0716e781f13be60a09840c795221 Mon Sep 17 00:00:00 2001 From: Joshua Kim <20001595+joshua-kim@users.noreply.github.com> Date: Thu, 18 Feb 2021 18:26:24 -0800 Subject: [PATCH] =?UTF-8?q?Fixing=20bug=20where=20ShardFilter=20parameter?= =?UTF-8?q?=20for=20ListShards=20was=20being=20passe=E2=80=A6=20(#788)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 --- .../amazon/kinesis/leases/KinesisShardDetector.java | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/KinesisShardDetector.java b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/KinesisShardDetector.java index 96a0de6a..189ba18b 100644 --- a/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/KinesisShardDetector.java +++ b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/KinesisShardDetector.java @@ -192,12 +192,16 @@ public class KinesisShardDetector implements ShardDetector { exceptionManager.add(ResourceInUseException.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)) { - request = request.streamName(streamIdentifier.streamName()); + builder = builder.streamName(streamIdentifier.streamName()).shardFilter(shardFilter); } 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; LimitExceededException lastException = null; int remainingRetries = maxListShardsRetryAttempts; @@ -205,7 +209,7 @@ public class KinesisShardDetector implements ShardDetector { while (result == null) { try { try { - result = getListShardsResponse(request.build()); + result = getListShardsResponse(request); } catch (ExecutionException e) { throw exceptionManager.apply(e.getCause()); } catch (InterruptedException e) {