docs: protected Supplier<Stream<Original>> supplierForSegments(ConsistentHash ch, IntSet targetSegments,
                                                                Set<Object> excludedKeys, boolean usePrimary) {
        if (!ch.getMembers().contains(localAddress)) {
           return Stream::empty;
        }
        IntSet segments;
        if (usePrimary) {
           if (targetSegments != null) {
              segments = IntSets.mutableCopyFrom(ch.getPrimarySegmentsForOwner(localAddress));
              segments.retainAll(targetSegments);
           } else {
              segments = IntSets.from(ch.getPrimarySegmentsForOwner(localAddress));
           }
        } else {
           segments = targetSegments;
        }
  
        return () -> {
           if (segments != null && segments.isEmpty()) {
              return Stream.empty();
           }
  
           CacheStream<Original> stream = supplier.get().filterKeySegments(segments);
           if (keysToFilter != null) {
              stream = stream.filterKeys(keysToFilter);
           }
           if (excludedKeys != null && !excludedKeys.isEmpty()) {
              return stream.filter(e -> !excludedKeys.contains(toKeyFunction == null ? e : toKeyFunction.apply(e)));
           }
           // Make sure the stream is set to be parallel or not
           return parallel ? stream.parallel() : stream.sequential();
        };
     }