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

    kinesis.stubSubscribeToShard("shard-000", eventWithRecords(3, 7));
    kinesis.stubSubscribeToShard("shard-000", eventsWithoutRecords(10, 10));

    kinesis.stubSubscribeToShard("shard-001", eventWithRecords(3));
    kinesis
        .stubSubscribeToShard("shard-001", eventWithRecords(3, 5))
        .failWith(SdkClientException.create("this is recoverable", new ReadTimeoutException()));
    kinesis.stubSubscribeToShard("shard-001", eventsWithoutRecords(8, 8));

    KinesisReaderCheckpoint initialCheckpoint =
        initialLatestCheckpoint(ImmutableList.of("shard-000", "shard-001"));

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

    PoolAssertion.assertPool(pool)
        .givesCheckPointedRecords(
            ShardAssertion.shard("shard-000")
                .gives(KinesisRecordView.generate("shard-000", 0, 10))
                .withLastCheckpointSequenceNumber(19),
            ShardAssertion.shard("shard-001")
                .gives(KinesisRecordView.generate("shard-001", 0, 8))
                .withLastCheckpointSequenceNumber(15));

    assertThat(kinesis.subscribeRequestsSeen())
        .containsExactlyInAnyOrder(
            subscribeLatest("shard-000"),
            subscribeLatest("shard-001"),
            subscribeAfterSeqNumber("shard-000", "2"),
            subscribeAfterSeqNumber("shard-001", "2"),
            subscribeAfterSeqNumber("shard-000", "9"),
            subscribeAfterSeqNumber("shard-001", "7"),
            subscribeAfterSeqNumber("shard-000", "19"),
            subscribeAfterSeqNumber("shard-001", "15"));
  }