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 poolReSubscribesWhenManyRecoverableErrorsOccur() throws Exception {
    kinesis = new EFOStubbedKinesisAsyncClient(1);

    for (int i = 0; i < 250; i++) {
      kinesis
          .stubSubscribeToShard("shard-000", eventsWithRecords(i, 1))
          .failWith(new ReadTimeoutException());
    }

    kinesis.stubSubscribeToShard("shard-000", eventsWithoutRecords(250, 3));

    kinesis
        .stubSubscribeToShard("shard-001", eventsWithRecords(333, 250))
        .failWith(SdkClientException.create("this is recoverable", new ReadTimeoutException()));

    kinesis.stubSubscribeToShard("shard-001", eventsWithoutRecords(583, 3));

    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, 250))
                .withLastCheckpointSequenceNumber(252),
            ShardAssertion.shard("shard-001")
                .gives(KinesisRecordView.generate("shard-001", 333, 250))
                .withLastCheckpointSequenceNumber(585));

    assertThat(kinesis.subscribeRequestsSeen().size()).isEqualTo(255);
  }