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())); } DynamicCheckpointGenerator(String streamName, StartingPoint startingPoint);  DynamicCheckpointGenerator(
      String streamName,
      StartingPoint startingPoint,
      StartingPointShardsFinder startingPointShardsFinder); @Override KinesisReaderCheckpoint generate(SimplifiedKinesisClient kinesis); @Override String toString();  }
@Test public void shouldMapAllShardsToCheckpoints() throws Exception { when(shard1.getShardId()).thenReturn("shard-01"); when(shard2.getShardId()).thenReturn("shard-02"); when(shard3.getShardId()).thenReturn("shard-03"); Set<Shard> shards = Sets.newHashSet(shard1, shard2, shard3); StartingPoint startingPoint = new StartingPoint(InitialPositionInStream.LATEST); when(startingPointShardsFinder.findShardsAtStartingPoint( kinesisClient, "stream", startingPoint)) .thenReturn(shards); DynamicCheckpointGenerator underTest = new DynamicCheckpointGenerator("stream", startingPoint, startingPointShardsFinder); KinesisReaderCheckpoint checkpoint = underTest.generate(kinesisClient); assertThat(checkpoint).hasSize(3); }
@Test public void shouldMapAllValidShardsToCheckpoints() throws Exception { when(shard1.getShardId()).thenReturn("shard-01"); when(shard2.getShardId()).thenReturn("shard-02"); when(shard3.getShardId()).thenReturn("shard-03"); String streamName = "stream"; Set<Shard> shards = Sets.newHashSet(shard1, shard2); StartingPoint startingPoint = new StartingPoint(InitialPositionInStream.LATEST); when(startingPointShardsFinder.findShardsAtStartingPoint( kinesisClient, "stream", startingPoint)) .thenReturn(shards); DynamicCheckpointGenerator underTest = new DynamicCheckpointGenerator(streamName, startingPoint, startingPointShardsFinder); KinesisReaderCheckpoint checkpoint = underTest.generate(kinesisClient); assertThat(checkpoint) .hasSize(2) .doesNotContain(new ShardCheckpoint(streamName, shard3.getShardId(), startingPoint)); }