docs: @Override
     public Iterator<R> iterator() {
        log.tracef("Distributed iterator invoked with rehash: %s", rehashAware);
        if (!rehashAware) {
           // Non rehash doesn't care about lost segments or completed ones
           CloseableIterator<R> closeableIterator = nonRehashRemoteIterator(dm.getReadConsistentHash(), segmentsToFilter,
                 null, IdentityPublisherDecorator.getInstance(), intermediateOperations);
           onClose(closeableIterator::close);
           return closeableIterator;
        } else {
           Iterable<IntermediateOperation> ops = iteratorOperation.prepareForIteration(intermediateOperations,
                 (Function) nonNullKeyFunction());
           CloseableIterator<R> closeableIterator;
           if (segmentCompletionListener != null && iteratorOperation != IteratorOperation.FLAT_MAP) {
              closeableIterator = new CompletionListenerRehashIterator<>(ops, segmentCompletionListener);
           } else {
              closeableIterator = new RehashIterator<>(ops);
           }
           onClose(closeableIterator::close);
           // This cast messes up generic checking, but that is okay
           Function<R, R> function = iteratorOperation.getFunction();
           if (function != null) {
              return new IteratorMapper<>(closeableIterator, function);
           } else {
              return closeableIterator;
           }
        }
     }