void start(Iterable<ShardCheckpoint> checkpoints) {
    LOG.info(
        "Pool {} - starting for stream {} consumer {}. Checkpoints = {}",
        poolId,
        read.getStreamName(),
        consumerArn,
        checkpoints);
    for (ShardCheckpoint shardCheckpoint : checkpoints) {
      checkState(
          !state.containsKey(shardCheckpoint.getShardId()),
          "Duplicate shard id %s",
          shardCheckpoint.getShardId());
      ShardState shardState =
          new ShardState(
              initShardSubscriber(shardCheckpoint), shardCheckpoint, watermarkPolicyFactory);
      state.put(shardCheckpoint.getShardId(), shardState);
    }
  }
@Test
  public void poolReSubscribesAndSkipsAllRecordsWithAtTimestampGreaterThanRecords()
      throws Exception {
    kinesis = new EFOStubbedKinesisAsyncClient(10);
    kinesis.stubSubscribeToShard("shard-000", eventWithRecords(3));
    kinesis.stubSubscribeToShard("shard-001", eventWithRecords(11, 3));

    Instant shard000ts = Instant.now().plus(Duration.standardHours(1));
    Instant shard001ts = Instant.now().plus(Duration.standardHours(1));

    KinesisReaderCheckpoint initialCheckpoint =
        new KinesisReaderCheckpoint(
            ImmutableList.of(
                tsCheckpoint("shard-000", shard000ts), tsCheckpoint("shard-001", shard001ts)));

    pool = new EFOShardSubscribersPool(readSpec, consumerArn, kinesis);
    pool.start(initialCheckpoint);

    PoolAssertion.assertPool(pool)
        .givesCheckPointedRecords(
            ShardAssertion.shard("shard-000").gives().withLastCheckpointSequenceNumber(2),
            ShardAssertion.shard("shard-001").gives().withLastCheckpointSequenceNumber(13));

    assertThat(kinesis.subscribeRequestsSeen())
        .containsExactlyInAnyOrder(
            subscribeAtTs("shard-000", shard000ts),
            subscribeAtTs("shard-001", shard001ts),
            subscribeAfterSeqNumber("shard-000", "2"),
            subscribeAfterSeqNumber("shard-001", "13"));
  }