void start() throws TransientKinesisException {
    ImmutableMap.Builder<String, ShardRecordsIterator> shardsMap = ImmutableMap.builder();
    for (ShardCheckpoint checkpoint : initialCheckpoint) {
      shardsMap.put(checkpoint.getShardId(), createShardIterator(kinesis, checkpoint));
    }
    shardIteratorsMap.set(shardsMap.build());
    if (!shardIteratorsMap.get().isEmpty()) {
      recordsQueue =
          new ArrayBlockingQueue<>(queueCapacityPerShard * shardIteratorsMap.get().size());
      String streamName = initialCheckpoint.getStreamName();
      startReadingShards(shardIteratorsMap.get().values(), streamName);
    } else {
      // There are no shards to handle when restoring from an empty checkpoint. Empty checkpoints
      // are generated when the last shard handled by this pool was closed
      recordsQueue = new ArrayBlockingQueue<>(1);
    }
  }
@Test
  public void shouldStopReadersPoolAlsoWhenExceptionsOccurDuringStopping() throws Exception {
    when(firstIterator.readNextBatch()).thenThrow(KinesisShardClosedException.class);
    when(firstIterator.findSuccessiveShardRecordIterators())
        .thenThrow(TransientKinesisException.class)
        .thenReturn(Collections.emptyList());

    shardReadersPool.start();

    verify(firstIterator, timeout(TIMEOUT_IN_MILLIS).times(2)).readNextBatch();
  }