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 poolReSubscribesAndReadsRecordsAfterCheckPoint() throws Exception {
    kinesis = new EFOStubbedKinesisAsyncClient(10);
    kinesis.stubSubscribeToShard("shard-000", eventWithRecords(3));
    kinesis.stubSubscribeToShard("shard-001", eventWithRecords(11, 3));

    KinesisReaderCheckpoint initialCheckpoint =
        new KinesisReaderCheckpoint(
            ImmutableList.of(
                afterCheckpoint("shard-000", "0"), afterCheckpoint("shard-001", "11")));

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

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

    assertThat(kinesis.subscribeRequestsSeen())
        .containsExactlyInAnyOrder(
            subscribeAtSeqNumber("shard-000", "0"),
            subscribeAtSeqNumber("shard-001", "11"),
            subscribeAfterSeqNumber("shard-000", "2"),
            subscribeAfterSeqNumber("shard-001", "13"));
  }