class DynamicCheckpointGenerator implements CheckpointGenerator {

  @Override
  public KinesisReaderCheckpoint generate(SimplifiedKinesisClient kinesis)
      throws TransientKinesisException {
    Set<Shard> shardsAtStartingPoint =
        startingPointShardsFinder.findShardsAtStartingPoint(kinesis, streamName, startingPoint);
    LOG.info(
        "Creating a checkpoint with following shards {} at {}",
        shardsAtStartingPoint,
        startingPoint.getTimestamp());
    return new KinesisReaderCheckpoint(
        shardsAtStartingPoint.stream()
            .map(shard -> new ShardCheckpoint(streamName, shard.getShardId(), startingPoint))
            .collect(Collectors.toList()));
  }

  public  DynamicCheckpointGenerator(String streamName, StartingPoint startingPoint);
  public  DynamicCheckpointGenerator(
      String streamName,
      StartingPoint startingPoint,
      StartingPointShardsFinder startingPointShardsFinder);

}