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 poolReSubscribesFromInitialWhenRecoverableErrorOccursImmediately() throws Exception {
    kinesis = new EFOStubbedKinesisAsyncClient(10);
    kinesis.stubSubscribeToShard("shard-000").failWith(new ReadTimeoutException());
    kinesis.stubSubscribeToShard("shard-000", eventWithRecords(550, 3));
    kinesis.stubSubscribeToShard("shard-000", eventsWithoutRecords(553, 1));

    KinesisReaderCheckpoint initialCheckpoint =
        initialLatestCheckpoint(ImmutableList.of("shard-000"));
    pool = new EFOShardSubscribersPool(readSpec, consumerArn, kinesis, 1);
    pool.start(initialCheckpoint);

    PoolAssertion.assertPool(pool)
        .givesCheckPointedRecords(
            ShardAssertion.shard("shard-000")
                .gives(KinesisRecordView.generate("shard-000", 550, 3))
                .withLastCheckpointSequenceNumber(553));

    assertThat(kinesis.subscribeRequestsSeen())
        .containsExactlyInAnyOrder(
            subscribeLatest("shard-000"),
            subscribeLatest("shard-000"),
            subscribeAfterSeqNumber("shard-000", "552"),
            subscribeAfterSeqNumber("shard-000", "553"));
  }