class StartingPointShardsFinder implements Serializable {

  Set<Shard> findShardsAtStartingPoint(
      SimplifiedKinesisClient kinesis, String streamName, StartingPoint startingPoint)
      throws TransientKinesisException {
    List<Shard> allShards = kinesis.listShards(streamName);
    Set<Shard> initialShards = findInitialShardsWithoutParents(streamName, allShards);

    Set<Shard> startingPointShards = new HashSet<>();
    Set<Shard> expiredShards;
    do {
      Set<Shard> validShards = validateShards(kinesis, initialShards, streamName, startingPoint);
      startingPointShards.addAll(validShards);
      expiredShards = Sets.difference(initialShards, validShards);
      if (!expiredShards.isEmpty()) {
        LOG.info(
            "Following shards expired for {} stream at '{}' starting point: {}",
            streamName,
            startingPoint,
            expiredShards);
      }
      initialShards = findNextShards(allShards, expiredShards);
    } while (!expiredShards.isEmpty());
    return startingPointShards;
  }

}