@Override
  public CloseableIterator<T> iterator() {
    ParallelIterator<T> iter =
        new ParallelIterator<>(iterables, workerPool, approximateMaxQueueSize);
    addCloseable(iter);
    return iter;
  }
@Test
  public void closeMoreDataParallelIteratorWithoutCompleteIteration() {
    ExecutorService executor = Executors.newFixedThreadPool(1);
    Iterator<Integer> integerIterator =
        new Iterator<Integer>() {
          private int number = 1;

          @Override
          public boolean hasNext() {
            if (number > 1000) {
              return false;
            }

            number++;
            return true;
          }

          @Override
          public Integer next() {
            try {
              // sleep to control number generate rate
              Thread.sleep(10);
            } catch (InterruptedException e) {
              // Sleep interrupted, we ignore it!
            }
            return number;
          }
        };
    Iterable<CloseableIterable<Integer>> transform =
        Iterables.transform(
            Lists.newArrayList(1),
            item ->
                new CloseableIterable<Integer>() {
                  @Override
                  public void close() {}

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

    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())
                    .as("Queue is not empty after cleaning")
                    .isEqualTo(0));
  }