docs: public <Original, E> OnCloseIterator<E> start(Address origin, Supplier<Stream<Original>> streamSupplier,
           Iterable<IntermediateOperation> intOps, Object requestId) {
        if (trace) {
           log.tracef("Iterator requested from %s using requestId %s", origin, requestId);
        }
        BaseStream stream = streamSupplier.get();
        for (IntermediateOperation intOp : intOps) {
           stream = intOp.perform(stream);
        }
  
        OnCloseIterator<E> iter = new IteratorCloser<>((CloseableIterator<E>) Closeables.iterator(stream));
        // When the iterator is closed make sure to clean up
        iter.onClose(() -> closeIterator(origin, requestId));
        currentRequests.put(requestId, iter);
        // This will be null if there have been no iterators for this node.
        // If the originating node died before we start this iterator this could be null as well. In this case the
        // iterator will be closed on the next view change.
        Set<Object> ids = ownerRequests.computeIfAbsent(origin, k -> ConcurrentHashMap.newKeySet());
        ids.add(requestId);
        return iter;
     }