@Override
  public KinesisReaderCheckpoint generate(SimplifiedKinesisClient kinesis)
      throws TransientKinesisException {
    List<Shard> streamShards = kinesis.listShardsAtPoint(streamName, startingPoint);
    LOG.info(
        "Creating a checkpoint with following shards {} at {}",
        streamShards,
        startingPoint.getTimestamp());
    return new KinesisReaderCheckpoint(
        streamShards.stream()
            .map(shard -> new ShardCheckpoint(streamName, shard.getShardId(), startingPoint))
            .collect(Collectors.toList()));
  }
@Test
  public void shouldMapAllShardsToCheckpoints() throws Exception {
    when(shard1.getShardId()).thenReturn("shard-01");
    when(shard2.getShardId()).thenReturn("shard-02");
    when(shard3.getShardId()).thenReturn("shard-03");
    List<Shard> shards = ImmutableList.of(shard1, shard2, shard3);
    String streamName = "stream";
    StartingPoint startingPoint = new StartingPoint(InitialPositionInStream.LATEST);
    when(kinesisClient.listShardsAtPoint(streamName, startingPoint)).thenReturn(shards);
    DynamicCheckpointGenerator underTest =
        new DynamicCheckpointGenerator(streamName, startingPoint);

    KinesisReaderCheckpoint checkpoint = underTest.generate(kinesisClient);

    assertThat(checkpoint).hasSize(3);
  }