class SimplifiedKinesisClient {

  public List<Shard> listShards(final String streamName) throws TransientKinesisException {
    return wrapExceptions(
        () -> {
          List<Shard> shards = Lists.newArrayList();
          String lastShardId = null;

          // DescribeStream has limits that can be hit fairly easily if we are attempting
          // to configure multiple KinesisIO inputs in the same account. Retry up to
          // LIST_SHARDS_DESCRIBE_STREAM_MAX_ATTEMPTS times if we end up hitting that limit.
          //
          // Only pass the wrapped exception up once that limit is reached. Use FluentBackoff
          // to implement the retry policy.
          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;
        });
  }

  public  SimplifiedKinesisClient(
      KinesisClient kinesis, CloudWatchClient cloudWatch, Integer limit);

}