SimplifiedKinesisClient { public List<Shard> listShards(final String streamName) throws TransientKinesisException { return wrapExceptions( () -> { List<Shard> shards = Lists.newArrayList(); String lastShardId = null; FluentBackoff retryBackoff = FluentBackoff.DEFAULT .withMaxRetries(LIST_SHARDS_DESCRIBE_STREAM_MAX_ATTEMPTS) .withInitialBackoff(LIST_SHARDS_DESCRIBE_STREAM_INITIAL_BACKOFF); StreamDescription description = null; do { BackOff backoff = retryBackoff.backoff(); Sleeper sleeper = Sleeper.DEFAULT; while (true) { try { description = kinesis .describeStream( DescribeStreamRequest.builder() .streamName(streamName) .exclusiveStartShardId(lastShardId) .build()) .streamDescription(); break; } catch (LimitExceededException exc) { if (!BackOffUtils.next(sleeper, backoff)) { throw exc; } } } shards.addAll(description.shards()); lastShardId = shards.get(shards.size() - 1).shardId(); } while (description.hasMoreShards()); return shards; }); } SimplifiedKinesisClient(
      KinesisClient kinesis, CloudWatchClient cloudWatch, Integer limit); static SimplifiedKinesisClient from(AWSClientsProvider provider, Integer limit); String getShardIterator(
      final String streamName,
      final String shardId,
      final ShardIteratorType shardIteratorType,
      final String startingSequenceNumber,
      final Instant timestamp); List<Shard> listShards(final String streamName); GetKinesisRecordsResult getRecords(String shardIterator, String streamName, String shardId); GetKinesisRecordsResult getRecords(
      final String shardIterator,
      final String streamName,
      final String shardId,
      final Integer limit); static List<KinesisClientRecord> deaggregate(List<Record> records); long getBacklogBytes(String streamName, Instant countSince); long getBacklogBytes(
      final String streamName, final Instant countSince, final Instant countTo);  }
@Test public void shouldListAllShards() throws Exception { Shard shard1 = Shard.builder().shardId(SHARD_1).build(); Shard shard2 = Shard.builder().shardId(SHARD_2).build(); Shard shard3 = Shard.builder().shardId(SHARD_3).build(); when(kinesis.describeStream( DescribeStreamRequest.builder().streamName(STREAM).exclusiveStartShardId(null).build())) .thenReturn( DescribeStreamResponse.builder() .streamDescription(s -> s.shards(shard1, shard2).hasMoreShards(true)) .build()); when(kinesis.describeStream( DescribeStreamRequest.builder() .streamName(STREAM) .exclusiveStartShardId(SHARD_2) .build())) .thenReturn( DescribeStreamResponse.builder() .streamDescription(s -> s.shards(shard3).hasMoreShards(false)) .build()); List<Shard> shards = underTest.listShards(STREAM); assertThat(shards).containsOnly(shard1, shard2, shard3); }