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 poolFailsWhenConsumerDoesNotExist() throws Exception {
    kinesis = new EFOStubbedKinesisAsyncClient(10);
    kinesis
        .stubSubscribeToShard("shard-000", eventWithRecords(3))
        .failWith(
            new CompletionException(
                "Err ...",
                ResourceNotFoundException.builder()
                    .cause(null)
                    .awsErrorDetails(
                        AwsErrorDetails.builder()
                            .serviceName("Kinesis")
                            .errorCode("ResourceNotFoundException")
                            .errorMessage("Consumer consumer-01 not found.")
                            .build())
                    .build()));

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

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

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

    pool = new EFOShardSubscribersPool(readSpec, consumerArn, kinesis);
    pool.start(initialCheckpoint);
    Throwable exception = assertThrows(IOException.class, () -> waitForRecords(pool, 10));
    assertThat(exception.getMessage())
        .isEqualTo("java.util.concurrent.CompletionException: Err ...");
    Throwable cause = exception.getCause().getCause();
    assertThat(cause).isInstanceOf(ResourceNotFoundException.class);
    assertThat(cause.getMessage())
        .isEqualTo(
            "Consumer consumer-01 not found. (Service: Kinesis, Status Code: 0, Request ID: null)");
  }