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(streamName, lastShardId).getStreamDescription(); break; } catch (LimitExceededException exc) { if (!BackOffUtils.next(sleeper, backoff)) { throw exc; } } } shards.addAll(description.getShards()); lastShardId = shards.get(shards.size() - 1).getShardId(); } while (description.getHasMoreShards()); return shards; }); } SimplifiedKinesisClient(
      AmazonKinesis kinesis, AmazonCloudWatch 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); 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 = new Shard().withShardId(SHARD_1); Shard shard2 = new Shard().withShardId(SHARD_2); Shard shard3 = new Shard().withShardId(SHARD_3); when(kinesis.describeStream(STREAM, null)) .thenReturn( new DescribeStreamResult() .withStreamDescription( new StreamDescription().withShards(shard1, shard2).withHasMoreShards(true))); when(kinesis.describeStream(STREAM, SHARD_2)) .thenReturn( new DescribeStreamResult() .withStreamDescription( new StreamDescription().withShards(shard3).withHasMoreShards(false))); List<Shard> shards = underTest.listShards(STREAM); assertThat(shards).containsOnly(shard1, shard2, shard3); }