ConsumingQueueIterator is not thread safe

[bjrke]

The code of ConsumingQueueIterator is not thread safe because it is not using an atomic operation in computeNext

we have this stacktrace once in our whole cluster

java.util.NoSuchElementException: null
    at java.util.AbstractQueue.remove(AbstractQueue.java:117)
    at com.google.common.collect.ConsumingQueueIterator.computeNext(ConsumingQueueIterator.java:44)
    at com.google.common.collect.AbstractIterator.tryToComputeNext(AbstractIterator.java:146)
    at com.google.common.collect.AbstractIterator.hasNext(AbstractIterator.java:141)
    ...

Instead of

  public T computeNext() {
    // TODO(b/192579700): Use a ternary once it no longer confuses our nullness checker.
    if (queue.isEmpty()) {
      return endOfData();
    }
    return queue.remove();
  }

I suggest sth like this but this changes the semantics since queued nulls mess this up... ofc its questionable if it makes sense at all

  protected T computeNext() {
    final var result = queue.poll();
    return result == null ? endOfData() : result;
  }

for me its also unclear why computeNext is public...

[dcm03]

The ConsumingQueueIterator class is an iterator that is designed to consume elements from a queue as they are encountered. The computeNext method is used to compute the next element in the iterator, by removing the element from the queue and returning it.

As you pointed out, the current implementation of the computeNext method is not thread-safe, because it is not using an atomic operation to remove the element from the queue. This means that if multiple threads are accessing the iterator concurrently, it is possible for one thread to remove an element from the queue just before another thread tries to remove the same element, causing the second thread to throw a NoSuchElementException.

To fix this issue, you could modify the computeNext method to use an atomic operation, such as poll, to remove the element from the queue. This would ensure that the element is removed from the queue in a thread-safe manner, preventing the NoSuchElementException from being thrown.

However, it's worth noting that this change would modify the semantics of the ConsumingQueueIterator, as poll does not throw an exception when the queue is empty, unlike remove. This means that the computeNext method would return null when the queue is empty, instead of throwing a NoSuchElementException.

It's also unclear why the computeNext method is public, as it is intended to be called by the iterator itself, rather than by external code. You may want to consider changing the visibility of the computeNext method to protected or private to reflect this.

[cpovirk]

This means that the computeNext method would return null when the queue is empty, instead of throwing a NoSuchElementException.

Thanks, yes, that's the reason. This might be nice to document, since users might reasonably assume thread-safety for a Queue.

It's also unclear why the computeNext method is public, as it is intended to be called by the iterator itself, rather than by external code. You may want to consider changing the visibility of the computeNext method to protected or private to reflect this.

That does sound like an accident. The method needs to be at least protected (since it overrides a protected abstract method), but there's no need for public. It looks like we did the same thing inside symmetricDifference.

[ayushsinghal90]

Raised a cr to contribute. [First contribution to the package 🙂].

[Grumpy2869]

The code of ConsumingQueueIterator is not thread safe because it is not using an atomic operation in computeNext

we have this stacktrace once in our whole clus
Duplicate of #ter

java.util.NoSuchElementException: null
    at java.util.AbstractQueue.remove(AbstractQueue.java:117)
    at com.google.common.collect.ConsumingQueueIterator.computeNext(ConsumingQueueIterator.java:44)
    at com.google.common.collect.AbstractIterator.tryToComputeNext(AbstractIterator.java:146)
    at com.google.common.collect.AbstractIterator.hasNext(AbstractIterator.java:141)
    ...

Instead of

  public T computeNext() {
    // TODO(b/192579700): Use a ternary once it no longer confuses our nullness checker.
    if (queue.isEmpty()) {
      return endOfData();
    }
    return queue.remove();
  }

I suggest sth like this but this changes the semantics since queued nulls mess this up... ofc its questionable if it makes sense at all

  protected T computeNext() {
    final var result = queue.poll();
    return result == null ? endOfData() : result;
  }

for me its also unclear why computeNext is public...

#6141 (comment)