@Override
  public CloseableIterator<T> iterator() {
    ParallelIterator<T> iter =
        new ParallelIterator<>(iterables, workerPool, approximateMaxQueueSize);
    addCloseable(iter);
    return iter;
  }
@Test
  public void closeParallelIteratorWithoutCompleteIteration() {
    ExecutorService executor = Executors.newFixedThreadPool(1);

    Iterable<CloseableIterable<Integer>> transform =
        Iterables.transform(
            Lists.newArrayList(1, 2, 3, 4, 5),
            item ->
                new CloseableIterable<Integer>() {
                  @Override
                  public void close() {}

                  @Override
                  public CloseableIterator<Integer> iterator() {
                    return CloseableIterator.withClose(Collections.singletonList(item).iterator());
                  }
                });

    ParallelIterable<Integer> parallelIterable = new ParallelIterable<>(transform, executor);
    ParallelIterator<Integer> iterator = (ParallelIterator<Integer>) parallelIterable.iterator();

    assertThat(iterator.hasNext()).isTrue();
    assertThat(iterator.next()).isNotNull();
    Awaitility.await("Queue is populated")
        .atMost(5, TimeUnit.SECONDS)
        .untilAsserted(() -> queueHasElements(iterator));
    iterator.close();
    Awaitility.await("Queue is cleared")
        .atMost(5, TimeUnit.SECONDS)
        .untilAsserted(() -> assertThat(iterator.queueSize()).isEqualTo(0));
  }