diff --git a/core/src/main/java/org/infinispan/CacheStream.java b/core/src/main/java/org/infinispan/CacheStream.java
index 6454358d85cb..29c9a1ed837f 100644
--- a/core/src/main/java/org/infinispan/CacheStream.java
+++ b/core/src/main/java/org/infinispan/CacheStream.java
@@ -217,6 +217,27 @@ interface SegmentCompletionListener {
     */
    void forEach(SerializableConsumer<? super R> action);
 
+   /**
+    * Same as {@link CacheStream#forEach(Consumer)} except that it takes a {@link BiConsumer} that provides access
+    * to the underlying {@link Cache} that is backing this stream.
+    * <p>
+    * Note that the <code>CacheAware</code> interface is not supported for injection using this method as the cache
+    * is provided in the consumer directly.
+    * @param action consumer to be ran for each element in the stream
+    * @param <K> key type of the cache
+    * @param <V> value type of the cache
+    */
+   <K, V> void forEach(BiConsumer<Cache<K, V>, ? super R> action);
+
+   /**
+    * Same as {@link CacheStream#forEach(BiConsumer)} except that the <code>BiConsumer</code> must also implement
+    * <code>Serializable</code>
+    * @param action consumer to be ran for each element in the stream
+    * @param <K> key type of the cache
+    * @param <V> value type of the cache
+    */
+   <K, V> void forEach(SerializableBiConsumer<Cache<K, V>, ? super R> action);
+
    /**
     * {@inheritDoc}
     * <p>Usage of this operator requires closing this stream after you are done with the iterator.  The preferred
diff --git a/core/src/main/java/org/infinispan/DoubleCacheStream.java b/core/src/main/java/org/infinispan/DoubleCacheStream.java
index 6de3bf444954..0cba19b562a4 100644
--- a/core/src/main/java/org/infinispan/DoubleCacheStream.java
+++ b/core/src/main/java/org/infinispan/DoubleCacheStream.java
@@ -1,5 +1,6 @@
 package org.infinispan;
 
+import org.infinispan.util.function.SerializableBiConsumer;
 import org.infinispan.util.function.SerializableDoubleBinaryOperator;
 import org.infinispan.util.function.SerializableDoubleConsumer;
 import org.infinispan.util.function.SerializableDoubleFunction;
@@ -8,7 +9,6 @@
 import org.infinispan.util.function.SerializableDoubleToLongFunction;
 import org.infinispan.util.function.SerializableDoubleUnaryOperator;
 import org.infinispan.util.function.SerializableObjDoubleConsumer;
-import org.infinispan.util.function.SerializableBiConsumer;
 import org.infinispan.util.function.SerializableSupplier;
 
 import java.util.OptionalDouble;
@@ -198,6 +198,27 @@ public interface DoubleCacheStream extends DoubleStream {
     */
    void forEach(SerializableDoubleConsumer action);
 
+   /**
+    * Same as {@link DoubleCacheStream#forEach(DoubleConsumer)} except that it takes an {@link ObjDoubleConsumer} that
+    * provides access to the underlying {@link Cache} that is backing this stream.
+    * <p>
+    * Note that the <code>CacheAware</code> interface is not supported for injection using this method as the cache
+    * is provided in the consumer directly.
+    * @param action consumer to be ran for each element in the stream
+    * @param <K> key type of the cache
+    * @param <V> value type of the cache
+    */
+   <K, V> void forEach(ObjDoubleConsumer<Cache<K, V>> action);
+
+   /**
+    * Same as {@link DoubleCacheStream#forEach(ObjDoubleConsumer)} except that the <code>BiConsumer</code> must also implement
+    * <code>Serializable</code>
+    * @param action consumer to be ran for each element in the stream
+    * @param <K> key type of the cache
+    * @param <V> value type of the cache
+    */
+   <K, V> void forEach(SerializableObjDoubleConsumer<Cache<K, V>> action);
+
    /**
     * Same as {@link DoubleCacheStream#reduce(double, DoubleBinaryOperator)} except that the DoubleBinaryOperator must
     * also implement <code>Serializable</code>
diff --git a/core/src/main/java/org/infinispan/IntCacheStream.java b/core/src/main/java/org/infinispan/IntCacheStream.java
index 5965149c05e7..9d57b7cf6632 100644
--- a/core/src/main/java/org/infinispan/IntCacheStream.java
+++ b/core/src/main/java/org/infinispan/IntCacheStream.java
@@ -198,6 +198,27 @@ public interface IntCacheStream extends IntStream {
     */
    void forEach(SerializableIntConsumer action);
 
+   /**
+    * Same as {@link IntCacheStream#forEach(IntConsumer)} except that it takes an {@link ObjIntConsumer} that
+    * provides access to the underlying {@link Cache} that is backing this stream.
+    * <p>
+    * Note that the <code>CacheAware</code> interface is not supported for injection using this method as the cache
+    * is provided in the consumer directly.
+    * @param action consumer to be ran for each element in the stream
+    * @param <K> key type of the cache
+    * @param <V> value type of the cache
+    */
+   <K, V> void forEach(ObjIntConsumer<Cache<K, V>> action);
+
+   /**
+    * Same as {@link IntCacheStream#forEach(ObjIntConsumer)} except that the <code>BiConsumer</code> must also implement
+    * <code>Serializable</code>
+    * @param action consumer to be ran for each element in the stream
+    * @param <K> key type of the cache
+    * @param <V> value type of the cache
+    */
+   <K, V> void forEach(SerializableObjIntConsumer<Cache<K, V>> action);
+
    /**
     * Same as {@link IntCacheStream#reduce(int, IntBinaryOperator)} except that the IntBinaryOperator
     * must also implement <code>Serializable</code>
diff --git a/core/src/main/java/org/infinispan/LongCacheStream.java b/core/src/main/java/org/infinispan/LongCacheStream.java
index 70847ff197e9..a0975fa53566 100644
--- a/core/src/main/java/org/infinispan/LongCacheStream.java
+++ b/core/src/main/java/org/infinispan/LongCacheStream.java
@@ -206,6 +206,27 @@ public interface LongCacheStream extends LongStream {
     */
    void forEach(SerializableLongConsumer action);
 
+   /**
+    * Same as {@link LongCacheStream#forEach(LongConsumer)} except that it takes an {@link ObjLongConsumer} that
+    * provides access to the underlying {@link Cache} that is backing this stream.
+    * <p>
+    * Note that the <code>CacheAware</code> interface is not supported for injection using this method as the cache
+    * is provided in the consumer directly.
+    * @param action consumer to be ran for each element in the stream
+    * @param <K> key type of the cache
+    * @param <V> value type of the cache
+    */
+   <K, V> void forEach(ObjLongConsumer<Cache<K, V>> action);
+
+   /**
+    * Same as {@link LongCacheStream#forEach(ObjLongConsumer)} except that the <code>BiConsumer</code> must also implement
+    * <code>Serializable</code>
+    * @param action consumer to be ran for each element in the stream
+    * @param <K> key type of the cache
+    * @param <V> value type of the cache
+    */
+   <K, V> void forEach(SerializableObjLongConsumer<Cache<K, V>> action);
+
    /**
     * Same as {@link LongCacheStream#reduce(long, LongBinaryOperator)} except that the LongBinaryOperator must
     * also implement Serializable.
diff --git a/core/src/main/java/org/infinispan/stream/impl/AbstractCacheStream.java b/core/src/main/java/org/infinispan/stream/impl/AbstractCacheStream.java
index 0b9064b55230..e1f2e4013695 100644
--- a/core/src/main/java/org/infinispan/stream/impl/AbstractCacheStream.java
+++ b/core/src/main/java/org/infinispan/stream/impl/AbstractCacheStream.java
@@ -43,9 +43,8 @@
  * primitive types.
  * @param <T> The type returned by the stream
  * @param <S> The stream interface
- * @param <T_CONS> The consumer for this stream
  */
-public abstract class AbstractCacheStream<T, S extends BaseStream<T, S>, S2 extends S, T_CONS> implements BaseStream<T, S> {
+public abstract class AbstractCacheStream<T, S extends BaseStream<T, S>, S2 extends S> implements BaseStream<T, S> {
    protected final Log log = LogFactory.getLog(getClass());
 
    protected final Queue<IntermediateOperation> intermediateOperations;
@@ -100,7 +99,7 @@ protected AbstractCacheStream(Address localAddress, boolean parallel, Distributi
       intermediateOperations = new ArrayDeque<>();
    }
 
-   protected AbstractCacheStream(AbstractCacheStream<T, S, S2, T_CONS> other) {
+   protected AbstractCacheStream(AbstractCacheStream<T, S, S2> other) {
       this.intermediateOperations = other.intermediateOperations;
       this.localIntermediateOperations = other.localIntermediateOperations;
       this.localAddress = other.localAddress;
@@ -234,12 +233,12 @@ public void close() {
       }
    }
 
-   <R> R performOperation(Function<S, ? extends R> function, boolean retryOnRehash, BinaryOperator<R> accumulator,
+   <R> R performOperation(Function<? super S2, ? extends R> function, boolean retryOnRehash, BinaryOperator<R> accumulator,
                           Predicate<? super R> earlyTerminatePredicate) {
       return performOperation(function, retryOnRehash, accumulator, earlyTerminatePredicate, true);
    }
 
-   <R> R performOperation(Function<S, ? extends R> function, boolean retryOnRehash, BinaryOperator<R> accumulator,
+   <R> R performOperation(Function<? super S2, ? extends R> function, boolean retryOnRehash, BinaryOperator<R> accumulator,
            Predicate<? super R> earlyTerminatePredicate, boolean ignoreSorting) {
       // These operations are not affected by sorting, only by distinct
       if (intermediateType.shouldUseIntermediate(!ignoreSorting && sorted, distinct)) {
@@ -254,7 +253,7 @@ <R> R performOperation(Function<S, ? extends R> function, boolean retryOnRehash,
       }
    }
 
-   <R> R performOperation(Function<S, ? extends R> function, ResultsAccumulator<R> remoteResults,
+   <R> R performOperation(Function<? super S2, ? extends R> function, ResultsAccumulator<R> remoteResults,
                           Predicate<? super R> earlyTerminatePredicate) {
       ConsistentHash ch = dm.getConsistentHash();
       TerminalOperation<R> op = new SingleRunOperation<>(intermediateOperations,
@@ -283,7 +282,7 @@ <R> R performOperation(Function<S, ? extends R> function, ResultsAccumulator<R>
       }
    }
 
-   <R> R performOperationRehashAware(Function<S, ? extends R> function, boolean retryOnRehash,
+   <R> R performOperationRehashAware(Function<? super S2, ? extends R> function, boolean retryOnRehash,
                                      ResultsAccumulator<R> remoteResults, Predicate<? super R> earlyTerminatePredicate) {
       Set<Integer> segmentsToProcess = segmentsToFilter;
       TerminalOperation<R> op;
@@ -353,10 +352,8 @@ <R> R performOperationRehashAware(Function<S, ? extends R> function, boolean ret
       return remoteResults.currentValue;
    }
 
-   abstract KeyTrackingTerminalOperation<Object, T, Object> getForEach(T_CONS consumer,
-           Supplier<Stream<CacheEntry>> supplier);
-
-   void performRehashForEach(T_CONS consumer) {
+   void performRehashKeyTrackingOperation(
+           Function<Supplier<Stream<CacheEntry>>, KeyTrackingTerminalOperation<Object, ? extends T, Object>> function) {
       final AtomicBoolean complete = new AtomicBoolean();
 
       ConsistentHash segmentInfoCH = dm.getReadConsistentHash();
@@ -380,7 +377,7 @@ void performRehashForEach(T_CONS consumer) {
             segments = null;
             excludedKeys = Collections.emptySet();
          }
-         KeyTrackingTerminalOperation<Object, T, Object> op = getForEach(consumer, supplierForSegments(ch,
+         KeyTrackingTerminalOperation<Object, ? extends T, Object> op = function.apply(supplierForSegments(ch,
                  segmentsToProcess, excludedKeys));
          op.handleInjection(registry);
          Object id = csm.remoteStreamOperationRehashAware(getParallelDistribution(), parallel, ch, segmentsToProcess,
@@ -782,7 +779,7 @@ public boolean shouldUseIntermediate(boolean sorted, boolean distinct) {
       }
    }
 
-   <R> R performIntermediateRemoteOperation(Function<S, ? extends R> function) {
+   <R> R performIntermediateRemoteOperation(Function<? super S2, ? extends R> function) {
       switch (intermediateType) {
          case OBJ:
             return performObjIntermediateRemoteOperation(function);
@@ -797,39 +794,39 @@ <R> R performIntermediateRemoteOperation(Function<S, ? extends R> function) {
       }
    }
 
-   <R> R performIntegerIntermediateRemoteOperation(Function<S, ? extends R> function) {
+   <R> R performIntegerIntermediateRemoteOperation(Function<? super S2, ? extends R> function) {
       // TODO: once we don't have to box for primitive iterators we can remove this copy
       Queue<IntermediateOperation> copyOperations = new ArrayDeque<>(localIntermediateOperations);
       PrimitiveIterator.OfInt iterator = new DistributedIntCacheStream(this).remoteIterator();
-      SingleRunOperation<R, T, S> op = new SingleRunOperation<>(copyOperations,
+      SingleRunOperation<R, T, S, S2> op = new SingleRunOperation<>(copyOperations,
               () -> StreamSupport.intStream(Spliterators.spliteratorUnknownSize(
                       iterator, Spliterator.CONCURRENT), parallel), function);
       return op.performOperation();
    }
 
-   <R> R performDoubleIntermediateRemoteOperation(Function<S, ? extends R> function) {
+   <R> R performDoubleIntermediateRemoteOperation(Function<? super S2, ? extends R> function) {
       // TODO: once we don't have to box for primitive iterators we can remove this copy
       Queue<IntermediateOperation> copyOperations = new ArrayDeque<>(localIntermediateOperations);
       PrimitiveIterator.OfDouble iterator = new DistributedDoubleCacheStream(this).remoteIterator();
-      SingleRunOperation<R, T, S> op = new SingleRunOperation<>(copyOperations,
+      SingleRunOperation<R, T, S, S2> op = new SingleRunOperation<>(copyOperations,
               () -> StreamSupport.doubleStream(Spliterators.spliteratorUnknownSize(
                       iterator, Spliterator.CONCURRENT), parallel), function);
       return op.performOperation();
    }
 
-   <R> R performLongIntermediateRemoteOperation(Function<S, ? extends R> function) {
+   <R> R performLongIntermediateRemoteOperation(Function<? super S2, ? extends R> function) {
       // TODO: once we don't have to box for primitive iterators we can remove this copy
       Queue<IntermediateOperation> copyOperations = new ArrayDeque<>(localIntermediateOperations);
       PrimitiveIterator.OfLong iterator = new DistributedLongCacheStream(this).remoteIterator();
-      SingleRunOperation<R, T, S> op = new SingleRunOperation<>(copyOperations,
+      SingleRunOperation<R, T, S, S2> op = new SingleRunOperation<>(copyOperations,
               () -> StreamSupport.longStream(Spliterators.spliteratorUnknownSize(
                       iterator, Spliterator.CONCURRENT), parallel), function);
       return op.performOperation();
    }
 
-   <R> R performObjIntermediateRemoteOperation(Function<S, ? extends R> function) {
+   <R> R performObjIntermediateRemoteOperation(Function<? super S2, ? extends R> function) {
       Iterator<Object> iterator = new DistributedCacheStream<>(this).remoteIterator();
-      SingleRunOperation<R, T, S> op = new SingleRunOperation<>(localIntermediateOperations,
+      SingleRunOperation<R, T, S, S2> op = new SingleRunOperation<>(localIntermediateOperations,
               () -> StreamSupport.stream(Spliterators.spliteratorUnknownSize(
                       iterator, Spliterator.CONCURRENT), parallel), function);
       return op.performOperation();
diff --git a/core/src/main/java/org/infinispan/stream/impl/AbstractDelegatingCacheStream.java b/core/src/main/java/org/infinispan/stream/impl/AbstractDelegatingCacheStream.java
index c3f4b5da3455..cf73fa125d7d 100644
--- a/core/src/main/java/org/infinispan/stream/impl/AbstractDelegatingCacheStream.java
+++ b/core/src/main/java/org/infinispan/stream/impl/AbstractDelegatingCacheStream.java
@@ -1,5 +1,6 @@
 package org.infinispan.stream.impl;
 
+import org.infinispan.Cache;
 import org.infinispan.CacheStream;
 import org.infinispan.util.function.SerializableBiConsumer;
 import org.infinispan.util.function.SerializableBiFunction;
@@ -100,7 +101,17 @@ public void forEach(Consumer<? super R> action) {
 
    @Override
    public void forEach(SerializableConsumer<? super R> action) {
-      forEach((Consumer<? super R>) action);
+      castStream(underlyingStream).forEach(action);
+   }
+
+   @Override
+   public <K, V> void forEach(BiConsumer<Cache<K, V>, ? super R> action) {
+      castStream(underlyingStream).forEach(action);
+   }
+
+   @Override
+   public <K, V> void forEach(SerializableBiConsumer<Cache<K, V>, ? super R> action) {
+      castStream(underlyingStream).forEach(action);
    }
 
    @Override
@@ -120,7 +131,7 @@ public <A> A[] toArray(IntFunction<A[]> generator) {
 
    @Override
    public <A> A[] toArray(SerializableIntFunction<A[]> generator) {
-      return toArray((IntFunction<A[]>) generator);
+      return underlyingStream.toArray(generator);
    }
 
    @Override
@@ -130,7 +141,7 @@ public R reduce(R identity, BinaryOperator<R> accumulator) {
 
    @Override
    public R reduce(R identity, SerializableBinaryOperator<R> accumulator) {
-      return reduce(identity, (BinaryOperator<R>) accumulator);
+      return castStream(underlyingStream).reduce(identity, accumulator);
    }
 
    @Override
@@ -140,7 +151,7 @@ public Optional<R> reduce(BinaryOperator<R> accumulator) {
 
    @Override
    public Optional<R> reduce(SerializableBinaryOperator<R> accumulator) {
-      return reduce((BinaryOperator<R>) accumulator);
+      return castStream(underlyingStream).reduce(accumulator);
    }
 
    @Override
@@ -150,7 +161,7 @@ public <U> U reduce(U identity, BiFunction<U, ? super R, U> accumulator, BinaryO
 
    @Override
    public <U> U reduce(U identity, SerializableBiFunction<U, ? super R, U> accumulator, SerializableBinaryOperator<U> combiner) {
-      return reduce(identity, (BiFunction<U, ? super R, U>) accumulator, combiner);
+      return castStream(underlyingStream).reduce(identity, accumulator, combiner);
    }
 
    @Override
@@ -160,7 +171,7 @@ public <R1> R1 collect(Supplier<R1> supplier, BiConsumer<R1, ? super R> accumula
 
    @Override
    public <R1> R1 collect(SerializableSupplier<R1> supplier, SerializableBiConsumer<R1, ? super R> accumulator, SerializableBiConsumer<R1, R1> combiner) {
-      return collect((Supplier<R1>) supplier, accumulator, combiner);
+      return castStream(underlyingStream).collect(supplier, accumulator, combiner);
    }
 
    @Override
@@ -221,7 +232,8 @@ public CacheStream<R> sorted(Comparator<? super R> comparator) {
 
    @Override
    public CacheStream<R> sorted(SerializableComparator<? super R> comparator) {
-      return sorted((Comparator<? super R>) comparator);
+      underlyingStream = castStream(underlyingStream).sorted(comparator);
+      return this;
    }
 
    @Override
@@ -232,7 +244,8 @@ public CacheStream<R> peek(Consumer<? super R> action) {
 
    @Override
    public CacheStream<R> peek(SerializableConsumer<? super R> action) {
-      return peek((Consumer<? super R>) action);
+      underlyingStream = castStream(underlyingStream).peek(action);
+      return this;
    }
 
    @Override
@@ -255,7 +268,8 @@ public CacheStream<R> filter(Predicate<? super R> predicate) {
 
    @Override
    public CacheStream<R> filter(SerializablePredicate<? super R> predicate) {
-      return filter((Predicate<? super R>) predicate);
+      underlyingStream = castStream(underlyingStream).filter(predicate);
+      return this;
    }
 
    @Override
@@ -266,7 +280,8 @@ public <R1> CacheStream<R1> map(Function<? super R, ? extends R1> mapper) {
 
    @Override
    public <R1> CacheStream<R1> map(SerializableFunction<? super R, ? extends R1> mapper) {
-      return map((Function<? super R, ? extends R1>) mapper);
+      underlyingStream = castStream(underlyingStream).map(mapper);
+      return (CacheStream<R1>) this;
    }
 
    @Override
@@ -277,7 +292,8 @@ public <R1> CacheStream<R1> flatMap(Function<? super R, ? extends Stream<? exten
 
    @Override
    public <R1> CacheStream<R1> flatMap(SerializableFunction<? super R, ? extends Stream<? extends R1>> mapper) {
-      return flatMap((Function<? super R, ? extends Stream<? extends R1>>) mapper);
+      underlyingStream = castStream(underlyingStream).flatMap(mapper);
+      return (CacheStream<R1>) this;
    }
 
    @Override
@@ -298,7 +314,7 @@ public Optional<R> min(Comparator<? super R> comparator) {
 
    @Override
    public Optional<R> min(SerializableComparator<? super R> comparator) {
-      return min((Comparator<? super R>) comparator);
+      return castStream(underlyingStream).min(comparator);
    }
 
    @Override
@@ -308,7 +324,7 @@ public Optional<R> max(Comparator<? super R> comparator) {
 
    @Override
    public Optional<R> max(SerializableComparator<? super R> comparator) {
-      return max((Comparator<? super R>) comparator);
+      return castStream(underlyingStream).max(comparator);
    }
 
    @Override
@@ -323,7 +339,7 @@ public boolean anyMatch(Predicate<? super R> predicate) {
 
    @Override
    public boolean anyMatch(SerializablePredicate<? super R> predicate) {
-      return anyMatch((Predicate<? super R>) predicate);
+      return castStream(underlyingStream).anyMatch(predicate);
    }
 
    @Override
@@ -333,7 +349,7 @@ public boolean allMatch(Predicate<? super R> predicate) {
 
    @Override
    public boolean allMatch(SerializablePredicate<? super R> predicate) {
-      return allMatch((Predicate<? super R>) predicate);
+      return castStream(underlyingStream).allMatch(predicate);
    }
 
    @Override
@@ -343,7 +359,7 @@ public boolean noneMatch(Predicate<? super R> predicate) {
 
    @Override
    public boolean noneMatch(SerializablePredicate<? super R> predicate) {
-      return noneMatch((Predicate<? super R>) predicate);
+      return castStream(underlyingStream).noneMatch(predicate);
    }
 
    @Override
diff --git a/core/src/main/java/org/infinispan/stream/impl/DistributedCacheStream.java b/core/src/main/java/org/infinispan/stream/impl/DistributedCacheStream.java
index 5e951fa85383..f1ed0938e9bc 100644
--- a/core/src/main/java/org/infinispan/stream/impl/DistributedCacheStream.java
+++ b/core/src/main/java/org/infinispan/stream/impl/DistributedCacheStream.java
@@ -1,5 +1,6 @@
 package org.infinispan.stream.impl;
 
+import org.infinispan.Cache;
 import org.infinispan.CacheStream;
 import org.infinispan.DoubleCacheStream;
 import org.infinispan.IntCacheStream;
@@ -16,6 +17,7 @@
 import org.infinispan.remoting.transport.Address;
 import org.infinispan.stream.impl.intops.object.*;
 import org.infinispan.stream.impl.termop.SingleRunOperation;
+import org.infinispan.stream.impl.termop.object.ForEachBiOperation;
 import org.infinispan.stream.impl.termop.object.ForEachOperation;
 import org.infinispan.stream.impl.termop.object.NoMapIteratorOperation;
 import org.infinispan.util.*;
@@ -60,7 +62,7 @@
  * nodes
  * @param <R> The type of the stream
  */
-public class DistributedCacheStream<R> extends AbstractCacheStream<R, Stream<R>, CacheStream<R>, Consumer<? super R>>
+public class DistributedCacheStream<R> extends AbstractCacheStream<R, Stream<R>, CacheStream<R>>
         implements CacheStream<R> {
 
    // This is a hack to allow for cast to work properly, since Java doesn't work as well with nested generics
@@ -506,9 +508,10 @@ public boolean noneMatch(SerializablePredicate<? super R> predicate) {
    public Optional<R> findFirst() {
       if (intermediateType.shouldUseIntermediate(sorted, distinct)) {
          Iterator<R> iterator = iterator();
-         SingleRunOperation<Optional<R>, R, Stream<R>> op = new SingleRunOperation<>(localIntermediateOperations,
+         SingleRunOperation<Optional<R>, R, Stream<R>, Stream<R>> op = new SingleRunOperation<>(localIntermediateOperations,
                  () -> StreamSupport.stream(Spliterators.spliteratorUnknownSize(
-                         iterator, Spliterator.CONCURRENT | Spliterator.NONNULL), parallel), Stream::findFirst);
+                         iterator, Spliterator.CONCURRENT | Spliterator.NONNULL), parallel),
+                 (Stream<R> r) -> r.findFirst());
          return op.performOperation();
       } else {
          return findAny();
@@ -533,7 +536,7 @@ public long count() {
    @Override
    public Iterator<R> iterator() {
       if (intermediateType.shouldUseIntermediate(sorted, distinct)) {
-         return performIntermediateRemoteOperation(Stream::iterator);
+         return performIntermediateRemoteOperation((Stream<R> s) -> s.iterator());
       } else {
          return remoteIterator();
       }
@@ -924,7 +927,8 @@ public void forEach(Consumer<? super R> action) {
       if (!rehashAware) {
          performOperation(TerminalFunctions.forEachFunction(action), false, (v1, v2) -> null, null);
       } else {
-         performRehashForEach(action);
+         performRehashKeyTrackingOperation(s -> new ForEachOperation<>(intermediateOperations, s, distributedBatchSize,
+                 action));
       }
    }
 
@@ -934,17 +938,27 @@ public void forEach(SerializableConsumer<? super R> action) {
    }
 
    @Override
-   KeyTrackingTerminalOperation getForEach(Consumer<? super R> consumer, Supplier<Stream<CacheEntry>> supplier) {
-      return new ForEachOperation<>(intermediateOperations, supplier, distributedBatchSize, consumer);
+   public <K, V> void forEach(BiConsumer<Cache<K, V>, ? super R> action) {
+      if (!rehashAware) {
+         performOperation(TerminalFunctions.forEachFunction(action), false, (v1, v2) -> null, null);
+      } else {
+         performRehashKeyTrackingOperation(s -> new ForEachBiOperation(intermediateOperations, s,
+                 distributedBatchSize, action));
+      }
+   }
+
+   @Override
+   public <K, V> void forEach(SerializableBiConsumer<Cache<K, V>, ? super R> action) {
+      forEach((BiConsumer<Cache<K, V>, ? super R>) action);
    }
 
    @Override
    public void forEachOrdered(Consumer<? super R> action) {
       if (sorted) {
          Iterator<R> iterator = iterator();
-         SingleRunOperation<Void, R,Stream<R>> op = new SingleRunOperation<>(localIntermediateOperations,
+         SingleRunOperation<Void, R,Stream<R>, Stream<R>> op = new SingleRunOperation<>(localIntermediateOperations,
                  () -> StreamSupport.stream(Spliterators.spliteratorUnknownSize(
-                         iterator, Spliterator.CONCURRENT | Spliterator.NONNULL), parallel), s -> {
+                         iterator, Spliterator.CONCURRENT | Spliterator.NONNULL), parallel), (Stream<R> s) -> {
             s.forEachOrdered(action);
             return null;
          });
diff --git a/core/src/main/java/org/infinispan/stream/impl/DistributedDoubleCacheStream.java b/core/src/main/java/org/infinispan/stream/impl/DistributedDoubleCacheStream.java
index 0e52f674968e..e0231212eadd 100644
--- a/core/src/main/java/org/infinispan/stream/impl/DistributedDoubleCacheStream.java
+++ b/core/src/main/java/org/infinispan/stream/impl/DistributedDoubleCacheStream.java
@@ -1,5 +1,6 @@
 package org.infinispan.stream.impl;
 
+import org.infinispan.Cache;
 import org.infinispan.CacheStream;
 import org.infinispan.DoubleCacheStream;
 import org.infinispan.IntCacheStream;
@@ -19,6 +20,8 @@
 import org.infinispan.stream.impl.intops.primitive.d.SortedDoubleOperation;
 import org.infinispan.stream.impl.termop.primitive.ForEachDoubleOperation;
 import org.infinispan.stream.impl.termop.primitive.ForEachFlatMapDoubleOperation;
+import org.infinispan.stream.impl.termop.primitive.ForEachFlatMapObjDoubleOperation;
+import org.infinispan.stream.impl.termop.primitive.ForEachObjDoubleOperation;
 import org.infinispan.util.function.SerializableDoubleBinaryOperator;
 import org.infinispan.util.function.SerializableDoubleConsumer;
 import org.infinispan.util.function.SerializableDoubleFunction;
@@ -57,7 +60,7 @@
  * class is only able to be created using {@link org.infinispan.CacheStream#mapToDouble(ToDoubleFunction)} or similar
  * methods from the {@link org.infinispan.CacheStream} interface.
  */
-public class DistributedDoubleCacheStream extends AbstractCacheStream<Double, DoubleStream, DoubleCacheStream, DoubleConsumer>
+public class DistributedDoubleCacheStream extends AbstractCacheStream<Double, DoubleStream, DoubleCacheStream>
         implements DoubleCacheStream {
    /**
     * This constructor is to be used only when a user calls a map or flat map method changing to a DoubleStream
@@ -191,7 +194,7 @@ public void forEach(DoubleConsumer action) {
       if (!rehashAware) {
          performOperation(TerminalFunctions.forEachFunction(action), false, (v1, v2) -> null, null);
       } else {
-         performRehashForEach(action);
+         performRehashKeyTrackingOperation(s -> getForEach(action, s));
       }
    }
 
@@ -201,6 +204,19 @@ public void forEach(SerializableDoubleConsumer action) {
    }
 
    @Override
+   public <K, V> void forEach(ObjDoubleConsumer<Cache<K, V>> action) {
+      if (!rehashAware) {
+         performOperation(TerminalFunctions.forEachFunction(action), false, (v1, v2) -> null, null);
+      } else {
+         performRehashKeyTrackingOperation(s -> getForEach(action, s));
+      }
+   }
+
+   @Override
+   public <K, V> void forEach(SerializableObjDoubleConsumer<Cache<K, V>> action) {
+      forEach((ObjDoubleConsumer<Cache<K, V>>) action);
+   }
+
    KeyTrackingTerminalOperation<Object, Double, Object> getForEach(DoubleConsumer consumer,
            Supplier<Stream<CacheEntry>> supplier) {
       if (iteratorOperation == IteratorOperation.FLAT_MAP) {
@@ -210,10 +226,19 @@ KeyTrackingTerminalOperation<Object, Double, Object> getForEach(DoubleConsumer c
       }
    }
 
+   <K, V> KeyTrackingTerminalOperation<Object, Double, Object> getForEach(ObjDoubleConsumer<Cache<K, V>> consumer,
+           Supplier<Stream<CacheEntry>> supplier) {
+      if (iteratorOperation == IteratorOperation.FLAT_MAP) {
+         return new ForEachFlatMapObjDoubleOperation(intermediateOperations, supplier, distributedBatchSize, consumer);
+      } else {
+         return new ForEachObjDoubleOperation(intermediateOperations, supplier, distributedBatchSize, consumer);
+      }
+   }
+
    @Override
    public void forEachOrdered(DoubleConsumer action) {
       if (intermediateType.shouldUseIntermediate(sorted, distinct)) {
-         performIntermediateRemoteOperation(s -> {
+         performIntermediateRemoteOperation((DoubleStream s) -> {
             s.forEachOrdered(action);
             return null;
          });
@@ -380,7 +405,7 @@ public boolean noneMatch(SerializableDoublePredicate predicate) {
    @Override
    public OptionalDouble findFirst() {
       if (intermediateType.shouldUseIntermediate(sorted, distinct)) {
-         return performIntermediateRemoteOperation(DoubleStream::findFirst);
+         return performIntermediateRemoteOperation((DoubleStream s) -> s.findFirst());
       } else {
          return findAny();
       }
diff --git a/core/src/main/java/org/infinispan/stream/impl/DistributedIntCacheStream.java b/core/src/main/java/org/infinispan/stream/impl/DistributedIntCacheStream.java
index e21136e3f13e..feb1a0762f68 100644
--- a/core/src/main/java/org/infinispan/stream/impl/DistributedIntCacheStream.java
+++ b/core/src/main/java/org/infinispan/stream/impl/DistributedIntCacheStream.java
@@ -1,5 +1,6 @@
 package org.infinispan.stream.impl;
 
+import org.infinispan.Cache;
 import org.infinispan.CacheStream;
 import org.infinispan.DoubleCacheStream;
 import org.infinispan.IntCacheStream;
@@ -20,7 +21,9 @@
 import org.infinispan.stream.impl.intops.primitive.i.SkipIntOperation;
 import org.infinispan.stream.impl.intops.primitive.i.SortedIntOperation;
 import org.infinispan.stream.impl.termop.primitive.ForEachFlatMapIntOperation;
+import org.infinispan.stream.impl.termop.primitive.ForEachFlatMapObjIntOperation;
 import org.infinispan.stream.impl.termop.primitive.ForEachIntOperation;
+import org.infinispan.stream.impl.termop.primitive.ForEachObjIntOperation;
 import org.infinispan.util.function.SerializableIntBinaryOperator;
 import org.infinispan.util.function.SerializableIntConsumer;
 import org.infinispan.util.function.SerializableIntPredicate;
@@ -60,7 +63,7 @@
  * class is only able to be created using {@link org.infinispan.CacheStream#mapToInt(ToIntFunction)} or similar
  * methods from the {@link org.infinispan.CacheStream} interface.
  */
-public class DistributedIntCacheStream extends AbstractCacheStream<Integer, IntStream, IntCacheStream, IntConsumer>
+public class DistributedIntCacheStream extends AbstractCacheStream<Integer, IntStream, IntCacheStream>
         implements IntCacheStream {
    /**
     * This constructor is to be used only when a user calls a map or flat map method changing to an IntStream
@@ -181,11 +184,6 @@ public IntCacheStream skip(long n) {
       return addIntermediateOperation(op);
    }
 
-   @Override
-   public void forEach(SerializableIntConsumer action) {
-      forEach((IntConsumer) action);
-   }
-
    @Override
    public LongCacheStream asLongStream() {
       addIntermediateOperationMap(AsLongIntOperation.getInstance());
@@ -212,11 +210,29 @@ public void forEach(IntConsumer action) {
       if (!rehashAware) {
          performOperation(TerminalFunctions.forEachFunction(action), false, (v1, v2) -> null, null);
       } else {
-         performRehashForEach(action);
+         performRehashKeyTrackingOperation(s -> getForEach(action, s));
       }
    }
 
    @Override
+   public void forEach(SerializableIntConsumer action) {
+      forEach((IntConsumer) action);
+   }
+
+   @Override
+   public <K, V> void forEach(ObjIntConsumer<Cache<K, V>> action) {
+      if (!rehashAware) {
+         performOperation(TerminalFunctions.forEachFunction(action), false, (v1, v2) -> null, null);
+      } else {
+         performRehashKeyTrackingOperation(s -> getForEach(action, s));
+      }
+   }
+
+   @Override
+   public <K, V> void forEach(SerializableObjIntConsumer<Cache<K, V>> action) {
+      forEach((ObjIntConsumer<Cache<K, V>>) action);
+   }
+
    KeyTrackingTerminalOperation<Object, Integer, Object> getForEach(IntConsumer consumer,
            Supplier<Stream<CacheEntry>> supplier) {
       if (iteratorOperation == IteratorOperation.FLAT_MAP) {
@@ -226,10 +242,19 @@ KeyTrackingTerminalOperation<Object, Integer, Object> getForEach(IntConsumer con
       }
    }
 
+   <K, V> KeyTrackingTerminalOperation<Object, Integer, Object> getForEach(ObjIntConsumer<Cache<K, V>> consumer,
+           Supplier<Stream<CacheEntry>> supplier) {
+      if (iteratorOperation == IteratorOperation.FLAT_MAP) {
+         return new ForEachFlatMapObjIntOperation(intermediateOperations, supplier, distributedBatchSize, consumer);
+      } else {
+         return new ForEachObjIntOperation(intermediateOperations, supplier, distributedBatchSize, consumer);
+      }
+   }
+
    @Override
    public void forEachOrdered(IntConsumer action) {
       if (intermediateType.shouldUseIntermediate(sorted, distinct)) {
-         performIntermediateRemoteOperation(s -> {
+         performIntermediateRemoteOperation((IntStream s) -> {
             s.forEachOrdered(action);
             return null;
          });
@@ -396,7 +421,7 @@ public boolean noneMatch(SerializableIntPredicate predicate) {
    @Override
    public OptionalInt findFirst() {
       if (intermediateType.shouldUseIntermediate(sorted, distinct)) {
-         return performIntermediateRemoteOperation(IntStream::findFirst);
+         return performIntermediateRemoteOperation((IntStream s) -> s.findFirst());
       } else {
          return findAny();
       }
diff --git a/core/src/main/java/org/infinispan/stream/impl/DistributedLongCacheStream.java b/core/src/main/java/org/infinispan/stream/impl/DistributedLongCacheStream.java
index d2ccc537c999..e1b03d3b4e2d 100644
--- a/core/src/main/java/org/infinispan/stream/impl/DistributedLongCacheStream.java
+++ b/core/src/main/java/org/infinispan/stream/impl/DistributedLongCacheStream.java
@@ -1,5 +1,6 @@
 package org.infinispan.stream.impl;
 
+import org.infinispan.Cache;
 import org.infinispan.CacheStream;
 import org.infinispan.DoubleCacheStream;
 import org.infinispan.IntCacheStream;
@@ -18,7 +19,9 @@
 import org.infinispan.stream.impl.intops.primitive.l.PeekLongOperation;
 import org.infinispan.stream.impl.intops.primitive.l.SortedLongOperation;
 import org.infinispan.stream.impl.termop.primitive.ForEachFlatMapLongOperation;
+import org.infinispan.stream.impl.termop.primitive.ForEachFlatMapObjLongOperation;
 import org.infinispan.stream.impl.termop.primitive.ForEachLongOperation;
+import org.infinispan.stream.impl.termop.primitive.ForEachObjLongOperation;
 import org.infinispan.util.function.SerializableLongBinaryOperator;
 import org.infinispan.util.function.SerializableLongConsumer;
 import org.infinispan.util.function.SerializableLongFunction;
@@ -58,7 +61,7 @@
  * class is only able to be created using {@link org.infinispan.CacheStream#mapToInt(ToIntFunction)} or similar
  * methods from the {@link org.infinispan.CacheStream} interface.
  */
-public class DistributedLongCacheStream extends AbstractCacheStream<Long, LongStream, LongCacheStream, LongConsumer>
+public class DistributedLongCacheStream extends AbstractCacheStream<Long, LongStream, LongCacheStream>
         implements LongCacheStream {
    /**
     * This constructor is to be used only when a user calls a map or flat map method changing to an IntStream
@@ -198,7 +201,7 @@ public void forEach(LongConsumer action) {
       if (!rehashAware) {
          performOperation(TerminalFunctions.forEachFunction(action), false, (v1, v2) -> null, null);
       } else {
-         performRehashForEach(action);
+         performRehashKeyTrackingOperation(s -> getForEach(action, s));
       }
    }
 
@@ -208,6 +211,19 @@ public void forEach(SerializableLongConsumer action) {
    }
 
    @Override
+   public <K, V> void forEach(ObjLongConsumer<Cache<K, V>> action) {
+      if (!rehashAware) {
+         performOperation(TerminalFunctions.forEachFunction(action), false, (v1, v2) -> null, null);
+      } else {
+         performRehashKeyTrackingOperation(s -> getForEach(action, s));
+      }
+   }
+
+   @Override
+   public <K, V> void forEach(SerializableObjLongConsumer<Cache<K, V>> action) {
+      forEach((ObjLongConsumer<Cache<K, V>>) action);
+   }
+
    KeyTrackingTerminalOperation<Object, Long, Object> getForEach(LongConsumer consumer,
            Supplier<Stream<CacheEntry>> supplier) {
       if (iteratorOperation == IteratorOperation.FLAT_MAP) {
@@ -217,10 +233,19 @@ KeyTrackingTerminalOperation<Object, Long, Object> getForEach(LongConsumer consu
       }
    }
 
+   <K, V> KeyTrackingTerminalOperation<Object, Long, Object> getForEach(ObjLongConsumer<Cache<K, V>> consumer,
+           Supplier<Stream<CacheEntry>> supplier) {
+      if (iteratorOperation == IteratorOperation.FLAT_MAP) {
+         return new ForEachFlatMapObjLongOperation(intermediateOperations, supplier, distributedBatchSize, consumer);
+      } else {
+         return new ForEachObjLongOperation(intermediateOperations, supplier, distributedBatchSize, consumer);
+      }
+   }
+
    @Override
    public void forEachOrdered(LongConsumer action) {
       if (intermediateType.shouldUseIntermediate(sorted, distinct)) {
-         performIntermediateRemoteOperation(s -> {
+         performIntermediateRemoteOperation((LongStream s) -> {
             s.forEachOrdered(action);
             return null;
          });
@@ -387,7 +412,7 @@ public boolean noneMatch(SerializableLongPredicate predicate) {
    @Override
    public OptionalLong findFirst() {
       if (intermediateType.shouldUseIntermediate(sorted, distinct)) {
-         return performIntermediateRemoteOperation(LongStream::findFirst);
+         return performIntermediateRemoteOperation((LongStream s) -> s.findFirst());
       } else {
          return findAny();
       }
diff --git a/core/src/main/java/org/infinispan/stream/impl/TerminalFunctions.java b/core/src/main/java/org/infinispan/stream/impl/TerminalFunctions.java
index cb444e37f220..6da3e0ebbe49 100644
--- a/core/src/main/java/org/infinispan/stream/impl/TerminalFunctions.java
+++ b/core/src/main/java/org/infinispan/stream/impl/TerminalFunctions.java
@@ -1,5 +1,10 @@
 package org.infinispan.stream.impl;
 
+import org.infinispan.Cache;
+import org.infinispan.CacheStream;
+import org.infinispan.DoubleCacheStream;
+import org.infinispan.IntCacheStream;
+import org.infinispan.LongCacheStream;
 import org.infinispan.commons.marshall.Externalizer;
 import org.infinispan.commons.marshall.SerializeWith;
 
@@ -130,18 +135,34 @@ public static <T> Function<Stream<T>, Void> forEachFunction(Consumer<? super T>
       return new ForEachFunction<>(consumer);
    }
 
+   public static <K, V, T> Function<CacheStream<T>, Void> forEachFunction(BiConsumer<Cache<K, V>, ? super T> consumer) {
+      return new ForEachBiConsumerFunction<>(consumer);
+   }
+
    public static Function<DoubleStream, Void> forEachFunction(DoubleConsumer consumer) {
       return new ForEachDoubleFunction<>(consumer);
    }
 
+   public static <K, V, T> Function<DoubleCacheStream, Void> forEachFunction(ObjDoubleConsumer<Cache<K, V>> consumer) {
+      return new ForEachObjDoubleFunction<>(consumer);
+   }
+
    public static Function<IntStream, Void> forEachFunction(IntConsumer consumer) {
       return new ForEachIntFunction<>(consumer);
    }
 
+   public static <K, V, T> Function<IntCacheStream, Void> forEachFunction(ObjIntConsumer<Cache<K, V>> consumer) {
+      return new ForEachObjIntFunction<>(consumer);
+   }
+
    public static Function<LongStream, Void> forEachFunction(LongConsumer consumer) {
       return new ForEachLongFunction<>(consumer);
    }
 
+   public static <K, V, T> Function<LongCacheStream, Void> forEachFunction(ObjLongConsumer<Cache<K, V>> consumer) {
+      return new ForEachObjLongFunction<>(consumer);
+   }
+
    public static <T> Function<Stream<T>, T> maxFunction(Comparator<? super T> comparator) {
       return new MaxFunction<>(comparator);
    }
@@ -1087,6 +1108,33 @@ public ForEachFunction readObject(ObjectInput input) throws IOException, ClassNo
       }
    }
 
+   @SerializeWith(value = ForEachBiConsumerFunction.ForEachBiConsumerFunctionExternalizer.class)
+   private static final class ForEachBiConsumerFunction<K, V, T> implements Function<CacheStream<T>, Void> {
+      private final BiConsumer<Cache<K, V>, ? super T> consumer;
+
+      private ForEachBiConsumerFunction(BiConsumer<Cache<K, V>, ? super T> consumer) {
+         this.consumer = consumer;
+      }
+
+      @Override
+      public Void apply(CacheStream<T> stream) {
+         stream.forEach(consumer);
+         return null;
+      }
+
+      public static final class ForEachBiConsumerFunctionExternalizer implements Externalizer<ForEachBiConsumerFunction> {
+         @Override
+         public void writeObject(ObjectOutput output, ForEachBiConsumerFunction object) throws IOException {
+            output.writeObject(object.consumer);
+         }
+
+         @Override
+         public ForEachBiConsumerFunction readObject(ObjectInput input) throws IOException, ClassNotFoundException {
+            return new ForEachBiConsumerFunction((BiConsumer) input.readObject());
+         }
+      }
+   }
+
    @SerializeWith(value = ForEachDoubleFunction.ForEachDoubleFunctionExternalizer.class)
    private static final class ForEachDoubleFunction<T> implements Function<DoubleStream, Void> {
       private final DoubleConsumer consumer;
@@ -1114,6 +1162,33 @@ public ForEachDoubleFunction readObject(ObjectInput input) throws IOException, C
       }
    }
 
+   @SerializeWith(value = ForEachObjDoubleFunction.ForEachObjDoubleFunctionExternalizer.class)
+   private static final class ForEachObjDoubleFunction<K, V> implements Function<DoubleCacheStream, Void> {
+      private final ObjDoubleConsumer<Cache<K, V>> consumer;
+
+      private ForEachObjDoubleFunction(ObjDoubleConsumer<Cache<K, V>> consumer) {
+         this.consumer = consumer;
+      }
+
+      @Override
+      public Void apply(DoubleCacheStream stream) {
+         stream.forEach(consumer);
+         return null;
+      }
+
+      public static final class ForEachObjDoubleFunctionExternalizer implements Externalizer<ForEachObjDoubleFunction> {
+         @Override
+         public void writeObject(ObjectOutput output, ForEachObjDoubleFunction object) throws IOException {
+            output.writeObject(object.consumer);
+         }
+
+         @Override
+         public ForEachObjDoubleFunction readObject(ObjectInput input) throws IOException, ClassNotFoundException {
+            return new ForEachObjDoubleFunction((ObjDoubleConsumer) input.readObject());
+         }
+      }
+   }
+
    @SerializeWith(value = ForEachIntFunction.ForEachIntFunctionExternalizer.class)
    private static final class ForEachIntFunction<T> implements Function<IntStream, Void> {
       private final IntConsumer consumer;
@@ -1141,6 +1216,33 @@ public ForEachIntFunction readObject(ObjectInput input) throws IOException, Clas
       }
    }
 
+   @SerializeWith(value = ForEachObjIntFunction.ForEachObjIntFunctionExternalizer.class)
+   private static final class ForEachObjIntFunction<K, V> implements Function<IntCacheStream, Void> {
+      private final ObjIntConsumer<Cache<K, V>> consumer;
+
+      private ForEachObjIntFunction(ObjIntConsumer<Cache<K, V>> consumer) {
+         this.consumer = consumer;
+      }
+
+      @Override
+      public Void apply(IntCacheStream stream) {
+         stream.forEach(consumer);
+         return null;
+      }
+
+      public static final class ForEachObjIntFunctionExternalizer implements Externalizer<ForEachObjIntFunction> {
+         @Override
+         public void writeObject(ObjectOutput output, ForEachObjIntFunction object) throws IOException {
+            output.writeObject(object.consumer);
+         }
+
+         @Override
+         public ForEachObjIntFunction readObject(ObjectInput input) throws IOException, ClassNotFoundException {
+            return new ForEachObjIntFunction((ObjIntConsumer) input.readObject());
+         }
+      }
+   }
+
    @SerializeWith(value = ForEachLongFunction.ForEachLongFunctionExternalizer.class)
    private static final class ForEachLongFunction<T> implements Function<LongStream, Void> {
       private final LongConsumer consumer;
@@ -1168,6 +1270,33 @@ public ForEachLongFunction readObject(ObjectInput input) throws IOException, Cla
       }
    }
 
+   @SerializeWith(value = ForEachObjLongFunction.ForEachObjLongFunctionExternalizer.class)
+   private static final class ForEachObjLongFunction<K, V> implements Function<LongCacheStream, Void> {
+      private final ObjLongConsumer<Cache<K, V>> consumer;
+
+      private ForEachObjLongFunction(ObjLongConsumer<Cache<K, V>> consumer) {
+         this.consumer = consumer;
+      }
+
+      @Override
+      public Void apply(LongCacheStream stream) {
+         stream.forEach(consumer);
+         return null;
+      }
+
+      public static final class ForEachObjLongFunctionExternalizer implements Externalizer<ForEachObjLongFunction> {
+         @Override
+         public void writeObject(ObjectOutput output, ForEachObjLongFunction object) throws IOException {
+            output.writeObject(object.consumer);
+         }
+
+         @Override
+         public ForEachObjLongFunction readObject(ObjectInput input) throws IOException, ClassNotFoundException {
+            return new ForEachObjLongFunction((ObjLongConsumer) input.readObject());
+         }
+      }
+   }
+
    @SerializeWith(value = MaxFunction.MaxFunctionExternalizer.class)
    private static final class MaxFunction<T> implements Function<Stream<T>, T> {
       private final Comparator<? super T> comparator;
diff --git a/core/src/main/java/org/infinispan/stream/impl/local/LocalCacheStream.java b/core/src/main/java/org/infinispan/stream/impl/local/LocalCacheStream.java
index 194eecabf217..14bcfe9100c1 100644
--- a/core/src/main/java/org/infinispan/stream/impl/local/LocalCacheStream.java
+++ b/core/src/main/java/org/infinispan/stream/impl/local/LocalCacheStream.java
@@ -248,6 +248,17 @@ public void forEach(SerializableConsumer<? super R> action) {
       forEach((Consumer<? super R>) action);
    }
 
+   @Override
+   public <K, V> void forEach(BiConsumer<Cache<K, V>, ? super R> action) {
+      Cache<K, V> cache = registry.getComponent(Cache.class);
+      createStream().forEach(e -> action.accept(cache, e));
+   }
+
+   @Override
+   public <K, V> void forEach(SerializableBiConsumer<Cache<K, V>, ? super R> action) {
+      forEach((BiConsumer<Cache<K, V>, ? super R>) action);
+   }
+
    @Override
    public void forEachOrdered(Consumer<? super R> action) {
       injectCache(action);
diff --git a/core/src/main/java/org/infinispan/stream/impl/local/LocalDoubleCacheStream.java b/core/src/main/java/org/infinispan/stream/impl/local/LocalDoubleCacheStream.java
index ee67681d81ca..5b122f337eeb 100644
--- a/core/src/main/java/org/infinispan/stream/impl/local/LocalDoubleCacheStream.java
+++ b/core/src/main/java/org/infinispan/stream/impl/local/LocalDoubleCacheStream.java
@@ -169,6 +169,17 @@ public void forEach(SerializableDoubleConsumer action) {
       forEach((DoubleConsumer) action);
    }
 
+   @Override
+   public <K, V> void forEach(ObjDoubleConsumer<Cache<K, V>> action) {
+      Cache<K, V> cache = registry.getComponent(Cache.class);
+      createStream().forEach(d -> action.accept(cache, d));
+   }
+
+   @Override
+   public <K, V> void forEach(SerializableObjDoubleConsumer<Cache<K, V>> action) {
+      forEach((ObjDoubleConsumer<Cache<K, V>>) action);
+   }
+
    @Override
    public void forEachOrdered(DoubleConsumer action) {
       injectCache(action);
diff --git a/core/src/main/java/org/infinispan/stream/impl/local/LocalIntCacheStream.java b/core/src/main/java/org/infinispan/stream/impl/local/LocalIntCacheStream.java
index 29f844d19894..e6fa9f4799e8 100644
--- a/core/src/main/java/org/infinispan/stream/impl/local/LocalIntCacheStream.java
+++ b/core/src/main/java/org/infinispan/stream/impl/local/LocalIntCacheStream.java
@@ -170,6 +170,17 @@ public void forEach(SerializableIntConsumer action) {
       forEach((IntConsumer) action);
    }
 
+   @Override
+   public <K, V> void forEach(ObjIntConsumer<Cache<K, V>> action) {
+      Cache<K, V> cache = registry.getComponent(Cache.class);
+      createStream().forEach(i -> action.accept(cache, i));
+   }
+
+   @Override
+   public <K, V> void forEach(SerializableObjIntConsumer<Cache<K, V>> action) {
+      forEach((ObjIntConsumer<Cache<K, V>>) action);
+   }
+
    @Override
    public void forEachOrdered(IntConsumer action) {
       injectCache(action);
diff --git a/core/src/main/java/org/infinispan/stream/impl/local/LocalLongCacheStream.java b/core/src/main/java/org/infinispan/stream/impl/local/LocalLongCacheStream.java
index 45a80b03134a..165f6505d7a8 100644
--- a/core/src/main/java/org/infinispan/stream/impl/local/LocalLongCacheStream.java
+++ b/core/src/main/java/org/infinispan/stream/impl/local/LocalLongCacheStream.java
@@ -170,6 +170,17 @@ public void forEach(SerializableLongConsumer action) {
       forEach((LongConsumer) action);
    }
 
+   @Override
+   public <K, V> void forEach(ObjLongConsumer<Cache<K, V>> action) {
+      Cache<K, V> cache = registry.getComponent(Cache.class);
+      createStream().forEach(l -> action.accept(cache, l));
+   }
+
+   @Override
+   public <K, V> void forEach(SerializableObjLongConsumer<Cache<K, V>> action) {
+      forEach((ObjLongConsumer<Cache<K, V>>) action);
+   }
+
    @Override
    public void forEachOrdered(LongConsumer action) {
       injectCache(action);
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/AbstractForEachOperation.java b/core/src/main/java/org/infinispan/stream/impl/termop/AbstractForEachOperation.java
index fd51ce59af77..0dafe139053a 100644
--- a/core/src/main/java/org/infinispan/stream/impl/termop/AbstractForEachOperation.java
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/AbstractForEachOperation.java
@@ -55,10 +55,10 @@ public List<V> performOperation(IntermediateCollector<Collection<V>> response) {
    @Override
    public Collection<CacheEntry<K, K>> performOperationRehashAware(
            IntermediateCollector<Collection<CacheEntry<K, K>>> response) {
-      // We only support sequential streams for iterator rehash aware
+      // We only support sequential streams for forEach rehash aware
       BaseStream<?, ?> stream = supplier.get().sequential();
 
-      List<CacheEntry<K, K>> collectedValues = new ArrayList(batchSize);
+      List<CacheEntry<K, K>> collectedValues = new ArrayList<>(batchSize);
 
       List<V> currentList = new ArrayList<>();
       ByRef<K> currentKey = new ByRef<>(null);
@@ -79,7 +79,7 @@ public Collection<CacheEntry<K, K>> performOperationRehashAware(
       }
 
       S convertedStream = ((S) stream);
-      // We rely on the fact that iterator processes 1 entry at a time
+      // We rely on the fact that forEach processes 1 entry at a time
       handleStreamForEach(convertedStream, currentList);
       if (!currentList.isEmpty()) {
          handleList(currentList);
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/SegmentRetryingOperation.java b/core/src/main/java/org/infinispan/stream/impl/termop/SegmentRetryingOperation.java
index fe192fa971fe..289b5e4bad54 100644
--- a/core/src/main/java/org/infinispan/stream/impl/termop/SegmentRetryingOperation.java
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/SegmentRetryingOperation.java
@@ -21,18 +21,18 @@
  * @param <T> type of the stream entries
  * @param <S> type of the stream itself
  */
-public class SegmentRetryingOperation<E, T, S extends BaseStream<T, S>> extends BaseTerminalOperation
+public class SegmentRetryingOperation<E, T, S extends BaseStream<T, S>, S2 extends S> extends BaseTerminalOperation
         implements TerminalOperation<E> {
    private static final Log log = LogFactory.getLog(SegmentRetryingOperation.class);
 
    private static final BaseStream<?, ?> EMPTY = Stream.empty();
 
-   private final Function<S, ? extends E> function;
+   private final Function<? super S2, ? extends E> function;
    private transient AtomicReference<BaseStream<?, ?>> streamRef = new AtomicReference<>(EMPTY);
    private transient AtomicBoolean continueTrying = new AtomicBoolean(true);
 
    public SegmentRetryingOperation(Iterable<IntermediateOperation> intermediateOperations,
-           Supplier<? extends Stream<?>> supplier, Function<S, ? extends E> function) {
+           Supplier<? extends Stream<?>> supplier, Function<? super S2, ? extends E> function) {
       super(intermediateOperations, supplier);
       this.function = function;
    }
@@ -65,7 +65,7 @@ private E innerPerformOperation(BaseStream<?, ?> stream) {
       for (IntermediateOperation intOp : intermediateOperations) {
          stream = intOp.perform(stream);
       }
-      return function.apply((S) stream);
+      return function.apply((S2) stream);
    }
 
    @Override
@@ -83,7 +83,7 @@ public E performOperation() {
       return keepTrying ? value : null;
    }
 
-   public Function<S, ? extends E> getFunction() {
+   public Function<? super S2, ? extends E> getFunction() {
       return function;
    }
 }
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/SingleRunOperation.java b/core/src/main/java/org/infinispan/stream/impl/termop/SingleRunOperation.java
index ce2a9ad0d980..e46696f6f277 100644
--- a/core/src/main/java/org/infinispan/stream/impl/termop/SingleRunOperation.java
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/SingleRunOperation.java
@@ -15,13 +15,13 @@
  * @param <R> type of the stream entries
  * @param <S> type of the stream itself
  */
-public class SingleRunOperation<E, R, S extends BaseStream<R, S>> extends BaseTerminalOperation
+public class SingleRunOperation<E, R, S extends BaseStream<R, S>, S2 extends S> extends BaseTerminalOperation
         implements TerminalOperation<E> {
-   private final Function<S, ? extends E> function;
+   private final Function<? super S2, ? extends E> function;
    private transient AtomicBoolean complete;
 
    public SingleRunOperation(Iterable<IntermediateOperation> intermediateOperations,
-                             Supplier<? extends BaseStream<?, ?>> supplier, Function<S, ? extends E> function) {
+                             Supplier<? extends BaseStream<?, ?>> supplier, Function<? super S2, ? extends E> function) {
       super(intermediateOperations, supplier);
       this.function = function;
       this.complete = new AtomicBoolean();
@@ -38,12 +38,12 @@ public E performOperation() {
       for (IntermediateOperation intOp : intermediateOperations) {
          stream = intOp.perform(stream);
       }
-      E value = function.apply((S) stream);
+      E value = function.apply((S2) stream);
       complete.set(true);
       return value;
    }
 
-   public Function<S, ? extends E> getFunction() {
+   public Function<? super S2, ? extends E> getFunction() {
       return function;
    }
 }
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/TerminalOperationExternalizer.java b/core/src/main/java/org/infinispan/stream/impl/termop/TerminalOperationExternalizer.java
index b0da25f99a60..1ffdd48c855a 100644
--- a/core/src/main/java/org/infinispan/stream/impl/termop/TerminalOperationExternalizer.java
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/TerminalOperationExternalizer.java
@@ -1,11 +1,13 @@
 package org.infinispan.stream.impl.termop;
 
+import org.infinispan.Cache;
 import org.infinispan.commons.io.UnsignedNumeric;
 import org.infinispan.commons.marshall.AdvancedExternalizer;
 import org.infinispan.commons.util.Util;
 import org.infinispan.marshall.core.Ids;
 import org.infinispan.stream.impl.intops.IntermediateOperation;
 import org.infinispan.stream.impl.termop.object.FlatMapIteratorOperation;
+import org.infinispan.stream.impl.termop.object.ForEachBiOperation;
 import org.infinispan.stream.impl.termop.object.ForEachOperation;
 import org.infinispan.stream.impl.termop.object.MapIteratorOperation;
 import org.infinispan.stream.impl.termop.object.NoMapIteratorOperation;
@@ -13,19 +15,29 @@
 import org.infinispan.stream.impl.termop.primitive.ForEachFlatMapDoubleOperation;
 import org.infinispan.stream.impl.termop.primitive.ForEachFlatMapIntOperation;
 import org.infinispan.stream.impl.termop.primitive.ForEachFlatMapLongOperation;
+import org.infinispan.stream.impl.termop.primitive.ForEachFlatMapObjDoubleOperation;
+import org.infinispan.stream.impl.termop.primitive.ForEachFlatMapObjIntOperation;
+import org.infinispan.stream.impl.termop.primitive.ForEachFlatMapObjLongOperation;
 import org.infinispan.stream.impl.termop.primitive.ForEachIntOperation;
 import org.infinispan.stream.impl.termop.primitive.ForEachLongOperation;
+import org.infinispan.stream.impl.termop.primitive.ForEachObjDoubleOperation;
+import org.infinispan.stream.impl.termop.primitive.ForEachObjIntOperation;
+import org.infinispan.stream.impl.termop.primitive.ForEachObjLongOperation;
 import org.jboss.marshalling.util.IdentityIntMap;
 
 import java.io.IOException;
 import java.io.ObjectInput;
 import java.io.ObjectOutput;
 import java.util.Set;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.DoubleConsumer;
 import java.util.function.Function;
 import java.util.function.IntConsumer;
 import java.util.function.LongConsumer;
+import java.util.function.ObjDoubleConsumer;
+import java.util.function.ObjIntConsumer;
+import java.util.function.ObjLongConsumer;
 
 /**
  * {@link AdvancedExternalizer} that provides functionality required for marshalling all of the various terminal
@@ -44,6 +56,13 @@ public class TerminalOperationExternalizer implements AdvancedExternalizer<BaseT
    private static final int FOREACH_DOUBLE = 9;
    private static final int FOREACH_INT = 10;
    private static final int FOREACH_LONG = 11;
+   private static final int FOREACH_BI = 12;
+   private static final int FOREACH_OBJ_DOUBLE = 13;
+   private static final int FOREACH_OBJ_INT = 14;
+   private static final int FOREACH_OBJ_LONG = 15;
+   private static final int FOREACH_FLAT_OBJ_DOUBLE = 16;
+   private static final int FOREACH_FLAT_OBJ_INT = 17;
+   private static final int FOREACH_FLAT_OBJ_LONG = 18;
 
    private final IdentityIntMap<Class<? extends BaseTerminalOperation>> operations = new IdentityIntMap<>();
 
@@ -60,6 +79,13 @@ public TerminalOperationExternalizer() {
       operations.put(ForEachDoubleOperation.class, FOREACH_DOUBLE);
       operations.put(ForEachIntOperation.class, FOREACH_INT);
       operations.put(ForEachLongOperation.class, FOREACH_LONG);
+      operations.put(ForEachBiOperation.class, FOREACH_BI);
+      operations.put(ForEachObjDoubleOperation.class, FOREACH_OBJ_DOUBLE);
+      operations.put(ForEachObjIntOperation.class, FOREACH_OBJ_INT);
+      operations.put(ForEachObjLongOperation.class, FOREACH_OBJ_LONG);
+      operations.put(ForEachFlatMapObjDoubleOperation.class, FOREACH_FLAT_OBJ_DOUBLE);
+      operations.put(ForEachFlatMapObjIntOperation.class, FOREACH_FLAT_OBJ_INT);
+      operations.put(ForEachFlatMapObjLongOperation.class, FOREACH_FLAT_OBJ_LONG);
    }
 
    @Override
@@ -68,7 +94,10 @@ public Set<Class<? extends BaseTerminalOperation>> getTypeClasses() {
               SegmentRetryingOperation.class, FlatMapIteratorOperation.class,
               MapIteratorOperation.class, NoMapIteratorOperation.class, ForEachOperation.class,
               ForEachFlatMapDoubleOperation.class, ForEachFlatMapIntOperation.class, ForEachFlatMapLongOperation.class,
-              ForEachDoubleOperation.class, ForEachIntOperation.class, ForEachLongOperation.class);
+              ForEachDoubleOperation.class, ForEachIntOperation.class, ForEachLongOperation.class,
+              ForEachBiOperation.class, ForEachObjDoubleOperation.class, ForEachObjIntOperation.class,
+              ForEachObjLongOperation.class, ForEachFlatMapObjDoubleOperation.class,
+              ForEachFlatMapObjIntOperation.class, ForEachFlatMapObjLongOperation.class);
    }
 
    @Override
@@ -125,6 +154,36 @@ public void writeObject(ObjectOutput output, BaseTerminalOperation object) throw
             UnsignedNumeric.writeUnsignedInt(output, ((ForEachLongOperation) object).getBatchSize());
             output.writeObject(((ForEachLongOperation) object).getConsumer());
             break;
+         case FOREACH_BI:
+            UnsignedNumeric.writeUnsignedInt(output, ((ForEachBiOperation) object).getBatchSize());
+            output.writeObject(((ForEachBiOperation) object).getConsumer());
+            break;
+         case FOREACH_OBJ_DOUBLE:
+            UnsignedNumeric.writeUnsignedInt(output, ((ForEachObjDoubleOperation) object).getBatchSize());
+            output.writeObject(((ForEachObjDoubleOperation) object).getConsumer());
+            break;
+         case FOREACH_OBJ_INT:
+            UnsignedNumeric.writeUnsignedInt(output, ((ForEachObjIntOperation) object).getBatchSize());
+            output.writeObject(((ForEachObjIntOperation<Object>) object).getConsumer());
+            break;
+         case FOREACH_OBJ_LONG:
+            UnsignedNumeric.writeUnsignedInt(output, ((ForEachObjLongOperation) object).getBatchSize());
+            output.writeObject(((ForEachObjLongOperation) object).getConsumer());
+            break;
+         case FOREACH_FLAT_OBJ_DOUBLE:
+            UnsignedNumeric.writeUnsignedInt(output, ((ForEachFlatMapObjDoubleOperation) object).getBatchSize());
+            output.writeObject(((ForEachFlatMapObjDoubleOperation) object).getConsumer());
+            break;
+         case FOREACH_FLAT_OBJ_INT:
+            UnsignedNumeric.writeUnsignedInt(output, ((ForEachFlatMapObjIntOperation) object).getBatchSize());
+            output.writeObject(((ForEachFlatMapObjIntOperation) object).getConsumer());
+            break;
+         case FOREACH_FLAT_OBJ_LONG:
+            UnsignedNumeric.writeUnsignedInt(output, ((ForEachFlatMapObjLongOperation) object).getBatchSize());
+            output.writeObject(((ForEachFlatMapObjLongOperation) object).getConsumer());
+            break;
+         default:
+            throw new IllegalArgumentException();
       }
    }
 
@@ -168,6 +227,27 @@ public BaseTerminalOperation readObject(ObjectInput input) throws IOException, C
          case FOREACH_LONG:
             return new ForEachLongOperation<>((Iterable<IntermediateOperation>) input.readObject(), null,
                     UnsignedNumeric.readUnsignedInt(input), (LongConsumer) input.readObject());
+         case FOREACH_BI:
+            return new ForEachBiOperation<>((Iterable<IntermediateOperation>) input.readObject(), null,
+                    UnsignedNumeric.readUnsignedInt(input), (BiConsumer) input.readObject());
+         case FOREACH_OBJ_DOUBLE:
+            return new ForEachObjDoubleOperation<>((Iterable<IntermediateOperation>) input.readObject(), null,
+                    UnsignedNumeric.readUnsignedInt(input), (ObjDoubleConsumer) input.readObject());
+         case FOREACH_OBJ_INT:
+            return new ForEachObjIntOperation<>((Iterable<IntermediateOperation>) input.readObject(), null,
+                    UnsignedNumeric.readUnsignedInt(input), (ObjIntConsumer) input.readObject());
+         case FOREACH_OBJ_LONG:
+            return new ForEachObjLongOperation<>((Iterable<IntermediateOperation>) input.readObject(), null,
+                    UnsignedNumeric.readUnsignedInt(input), (ObjLongConsumer) input.readObject());
+         case FOREACH_FLAT_OBJ_DOUBLE:
+            return new ForEachFlatMapObjDoubleOperation<>((Iterable<IntermediateOperation>) input.readObject(), null,
+                    UnsignedNumeric.readUnsignedInt(input), (ObjDoubleConsumer) input.readObject());
+         case FOREACH_FLAT_OBJ_INT:
+            return new ForEachFlatMapObjIntOperation<>((Iterable<IntermediateOperation>) input.readObject(), null,
+                    UnsignedNumeric.readUnsignedInt(input), (ObjIntConsumer) input.readObject());
+         case FOREACH_FLAT_OBJ_LONG:
+            return new ForEachFlatMapObjLongOperation<>((Iterable<IntermediateOperation>) input.readObject(), null,
+                    UnsignedNumeric.readUnsignedInt(input), (ObjLongConsumer) input.readObject());
 
          default:
             throw new IllegalArgumentException("Found invalid number " + number);
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/object/ForEachBiOperation.java b/core/src/main/java/org/infinispan/stream/impl/termop/object/ForEachBiOperation.java
new file mode 100644
index 000000000000..857adf19c868
--- /dev/null
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/object/ForEachBiOperation.java
@@ -0,0 +1,50 @@
+package org.infinispan.stream.impl.termop.object;
+
+import org.infinispan.Cache;
+import org.infinispan.container.entries.CacheEntry;
+import org.infinispan.factories.ComponentRegistry;
+import org.infinispan.stream.CacheAware;
+import org.infinispan.stream.impl.intops.IntermediateOperation;
+import org.infinispan.stream.impl.termop.AbstractForEachOperation;
+
+import java.util.List;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
+
+/**
+ * Terminal operation that handles for each where no map operations are defined
+ * @param <K> key type of the supplied stream
+ * @param <V> resulting value type
+ */
+public class ForEachBiOperation<K, V> extends AbstractForEachOperation<K, V, Stream<V>> {
+   private final BiConsumer<Cache<K, ?>, ? super V> consumer;
+   private transient Cache<K, ?> cache;
+
+   public ForEachBiOperation(Iterable<IntermediateOperation> intermediateOperations,
+           Supplier<Stream<CacheEntry>> supplier, int batchSize, BiConsumer<Cache<K, ?>, ? super V> consumer) {
+      super(intermediateOperations, supplier, batchSize);
+      this.consumer = consumer;
+   }
+
+   @Override
+   protected void handleList(List<V> list) {
+      list.forEach(e -> consumer.accept(cache, e));
+   }
+
+   @Override
+   protected void handleStreamForEach(Stream<V> stream, List<V> list) {
+      stream.forEach(list::add);
+   }
+
+   public BiConsumer<Cache<K, ?>, ? super V> getConsumer() {
+      return consumer;
+   }
+
+   @Override
+   public void handleInjection(ComponentRegistry registry) {
+      super.handleInjection(registry);
+      cache = registry.getComponent(Cache.class);
+   }
+}
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/primitive/AbstractForEachDoubleOperation.java b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/AbstractForEachDoubleOperation.java
new file mode 100644
index 000000000000..df4ee30da013
--- /dev/null
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/AbstractForEachDoubleOperation.java
@@ -0,0 +1,90 @@
+package org.infinispan.stream.impl.termop.primitive;
+
+import org.infinispan.container.entries.CacheEntry;
+import org.infinispan.container.entries.ImmortalCacheEntry;
+import org.infinispan.stream.impl.KeyTrackingTerminalOperation;
+import org.infinispan.stream.impl.intops.IntermediateOperation;
+import org.infinispan.stream.impl.termop.BaseTerminalOperation;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+import java.util.stream.BaseStream;
+import java.util.stream.DoubleStream;
+import java.util.stream.Stream;
+
+/**
+ * Terminal rehash aware operation that handles for each where no flat map operations are defined on a
+ * {@link DoubleStream}. Note this means it is an implied map intermediate operation.
+ * @param <K> key type of the supplied stream
+ */
+public abstract class AbstractForEachDoubleOperation<K> extends BaseTerminalOperation implements KeyTrackingTerminalOperation<K, Double, K> {
+   private final int batchSize;
+
+   public AbstractForEachDoubleOperation(Iterable<IntermediateOperation> intermediateOperations,
+           Supplier<Stream<CacheEntry>> supplier, int batchSize) {
+      super(intermediateOperations, supplier);
+      this.batchSize = batchSize;
+   }
+
+   @Override
+   public boolean lostSegment(boolean stopIfLost) {
+      // TODO: stop this early
+      return true;
+   }
+
+   @Override
+   public List<Double> performOperation(IntermediateCollector<Collection<Double>> response) {
+      /**
+       * This is for rehash only! {@link org.infinispan.stream.impl.termop.SingleRunOperation} should always be used for
+       * non rehash
+       */
+      throw new UnsupportedOperationException();
+   }
+
+   protected abstract void handleArray(double[] array, int size);
+
+   @Override
+   public Collection<CacheEntry<K, K>> performOperationRehashAware(
+           IntermediateCollector<Collection<CacheEntry<K, K>>> response) {
+      // We only support sequential streams for iterator rehash aware
+      BaseStream<?, ?> stream = supplier.get().sequential();
+
+      List<CacheEntry<K, K>> collectedValues = new ArrayList<>(batchSize);
+
+      double[] list = new double[batchSize];
+      AtomicInteger offset = new AtomicInteger();
+      Object[] currentKey = new Object[1];
+      stream = ((Stream<Map.Entry<K, ?>>) stream).peek(e -> {
+         if (offset.get() > 0) {
+            collectedValues.add(new ImmortalCacheEntry(currentKey[0], currentKey[0]));
+            if (collectedValues.size() >= batchSize) {
+               handleArray(list, offset.get());
+               response.sendDataResonse(collectedValues);
+               collectedValues.clear();
+               offset.set(0);
+            }
+         }
+         currentKey[0] = e.getKey();
+      });
+      for (IntermediateOperation intermediateOperation : intermediateOperations) {
+         stream = intermediateOperation.perform(stream);
+      }
+
+      DoubleStream convertedStream = ((DoubleStream)stream);
+      // We rely on the fact that iterator processes 1 entry at a time when sequential
+      convertedStream.forEach(d -> list[offset.getAndIncrement()] = d);
+      if (offset.get() > 0) {
+         handleArray(list, offset.get());
+         collectedValues.add(new ImmortalCacheEntry(currentKey[0], currentKey[0]));
+      }
+      return collectedValues;
+   }
+
+   public int getBatchSize() {
+      return batchSize;
+   }
+}
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/primitive/AbstractForEachIntOperation.java b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/AbstractForEachIntOperation.java
new file mode 100644
index 000000000000..5cd3d08dd851
--- /dev/null
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/AbstractForEachIntOperation.java
@@ -0,0 +1,90 @@
+package org.infinispan.stream.impl.termop.primitive;
+
+import org.infinispan.container.entries.CacheEntry;
+import org.infinispan.container.entries.ImmortalCacheEntry;
+import org.infinispan.stream.impl.KeyTrackingTerminalOperation;
+import org.infinispan.stream.impl.intops.IntermediateOperation;
+import org.infinispan.stream.impl.termop.BaseTerminalOperation;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+import java.util.stream.BaseStream;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+/**
+ * Terminal rehash aware operation that handles for each where no flat map operations are defined on a
+ * {@link IntStream}. Note this means it is an implied map intermediate operation.
+ * @param <K> key type of the supplied stream
+ */
+public abstract class AbstractForEachIntOperation<K> extends BaseTerminalOperation implements KeyTrackingTerminalOperation<K, Integer, K> {
+   private final int batchSize;
+
+   public AbstractForEachIntOperation(Iterable<IntermediateOperation> intermediateOperations,
+           Supplier<Stream<CacheEntry>> supplier, int batchSize) {
+      super(intermediateOperations, supplier);
+      this.batchSize = batchSize;
+   }
+
+   @Override
+   public boolean lostSegment(boolean stopIfLost) {
+      // TODO: stop this early
+      return true;
+   }
+
+   @Override
+   public List<Integer> performOperation(IntermediateCollector<Collection<Integer>> response) {
+      /**
+       * This is for rehash only! {@link org.infinispan.stream.impl.termop.SingleRunOperation} should always be used for
+       * non rehash
+       */
+      throw new UnsupportedOperationException();
+   }
+
+   protected abstract void handleArray(int[] array, int size);
+
+   @Override
+   public Collection<CacheEntry<K, K>> performOperationRehashAware(
+           IntermediateCollector<Collection<CacheEntry<K, K>>> response) {
+      // We only support sequential streams for iterator rehash aware
+      BaseStream<?, ?> stream = supplier.get().sequential();
+
+      List<CacheEntry<K, K>> collectedValues = new ArrayList<>(batchSize);
+
+      int[] list = new int[batchSize];
+      AtomicInteger offset = new AtomicInteger();
+      Object[] currentKey = new Object[1];
+      stream = ((Stream<Map.Entry<K, ?>>) stream).peek(e -> {
+         if (offset.get() > 0) {
+            collectedValues.add(new ImmortalCacheEntry(currentKey[0], currentKey[0]));
+            if (collectedValues.size() >= batchSize) {
+               handleArray(list, offset.get());
+               response.sendDataResonse(collectedValues);
+               collectedValues.clear();
+               offset.set(0);
+            }
+         }
+         currentKey[0] = e.getKey();
+      });
+      for (IntermediateOperation intermediateOperation : intermediateOperations) {
+         stream = intermediateOperation.perform(stream);
+      }
+
+      IntStream convertedStream = ((IntStream)stream);
+      // We rely on the fact that iterator processes 1 entry at a time when sequential
+      convertedStream.forEach(d -> list[offset.getAndIncrement()] = d);
+      if (offset.get() > 0) {
+         handleArray(list, offset.get());
+         collectedValues.add(new ImmortalCacheEntry(currentKey[0], currentKey[0]));
+      }
+      return collectedValues;
+   }
+
+   public int getBatchSize() {
+      return batchSize;
+   }
+}
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/primitive/AbstractForEachLongOperation.java b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/AbstractForEachLongOperation.java
new file mode 100644
index 000000000000..034849de6430
--- /dev/null
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/AbstractForEachLongOperation.java
@@ -0,0 +1,90 @@
+package org.infinispan.stream.impl.termop.primitive;
+
+import org.infinispan.container.entries.CacheEntry;
+import org.infinispan.container.entries.ImmortalCacheEntry;
+import org.infinispan.stream.impl.KeyTrackingTerminalOperation;
+import org.infinispan.stream.impl.intops.IntermediateOperation;
+import org.infinispan.stream.impl.termop.BaseTerminalOperation;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+import java.util.stream.BaseStream;
+import java.util.stream.LongStream;
+import java.util.stream.Stream;
+
+/**
+ * Terminal rehash aware operation that handles for each where no flat map operations are defined on a
+ * {@link LongStream}. Note this means it is an implied map intermediate operation.
+ * @param <K> key type of the supplied stream
+ */
+public abstract class AbstractForEachLongOperation<K> extends BaseTerminalOperation implements KeyTrackingTerminalOperation<K, Long, K> {
+   private final int batchSize;
+
+   public AbstractForEachLongOperation(Iterable<IntermediateOperation> intermediateOperations,
+           Supplier<Stream<CacheEntry>> supplier, int batchSize) {
+      super(intermediateOperations, supplier);
+      this.batchSize = batchSize;
+   }
+
+   @Override
+   public boolean lostSegment(boolean stopIfLost) {
+      // TODO: stop this early
+      return true;
+   }
+
+   @Override
+   public List<Long> performOperation(IntermediateCollector<Collection<Long>> response) {
+      /**
+       * This is for rehash only! {@link org.infinispan.stream.impl.termop.SingleRunOperation} should always be used for
+       * non rehash
+       */
+      throw new UnsupportedOperationException();
+   }
+
+   protected abstract void handleArray(long[] array, int size);
+
+   @Override
+   public Collection<CacheEntry<K, K>> performOperationRehashAware(
+           IntermediateCollector<Collection<CacheEntry<K, K>>> response) {
+      // We only support sequential streams for iterator rehash aware
+      BaseStream<?, ?> stream = supplier.get().sequential();
+
+      List<CacheEntry<K, K>> collectedValues = new ArrayList<>(batchSize);
+
+      long[] list = new long[batchSize];
+      AtomicInteger offset = new AtomicInteger();
+      Object[] currentKey = new Object[1];
+      stream = ((Stream<Map.Entry<K, ?>>) stream).peek(e -> {
+         if (offset.get() > 0) {
+            collectedValues.add(new ImmortalCacheEntry(currentKey[0], currentKey[0]));
+            if (collectedValues.size() >= batchSize) {
+               handleArray(list, offset.get());
+               response.sendDataResonse(collectedValues);
+               collectedValues.clear();
+               offset.set(0);
+            }
+         }
+         currentKey[0] = e.getKey();
+      });
+      for (IntermediateOperation intermediateOperation : intermediateOperations) {
+         stream = intermediateOperation.perform(stream);
+      }
+
+      LongStream convertedStream = ((LongStream)stream);
+      // We rely on the fact that iterator processes 1 entry at a time when sequential
+      convertedStream.forEach(d -> list[offset.getAndIncrement()] = d);
+      if (offset.get() > 0) {
+         handleArray(list, offset.get());
+         collectedValues.add(new ImmortalCacheEntry(currentKey[0], currentKey[0]));
+      }
+      return collectedValues;
+   }
+
+   public int getBatchSize() {
+      return batchSize;
+   }
+}
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachDoubleOperation.java b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachDoubleOperation.java
index 469efc807c6e..2b0e516e74fe 100644
--- a/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachDoubleOperation.java
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachDoubleOperation.java
@@ -2,21 +2,12 @@
 
 import org.infinispan.Cache;
 import org.infinispan.container.entries.CacheEntry;
-import org.infinispan.container.entries.ImmortalCacheEntry;
 import org.infinispan.factories.ComponentRegistry;
-import org.infinispan.stream.impl.KeyTrackingTerminalOperation;
-import org.infinispan.stream.impl.intops.IntermediateOperation;
-import org.infinispan.stream.impl.termop.BaseTerminalOperation;
 import org.infinispan.stream.CacheAware;
+import org.infinispan.stream.impl.intops.IntermediateOperation;
 
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.DoubleConsumer;
 import java.util.function.Supplier;
-import java.util.stream.BaseStream;
 import java.util.stream.DoubleStream;
 import java.util.stream.Stream;
 
@@ -25,75 +16,20 @@
  * {@link DoubleStream}. Note this means it is an implied map intermediate operation.
  * @param <K> key type of the supplied stream
  */
-public class ForEachDoubleOperation<K> extends BaseTerminalOperation implements KeyTrackingTerminalOperation<K, Double, K> {
-   private final int batchSize;
+public class ForEachDoubleOperation<K> extends AbstractForEachDoubleOperation<K> {
    private final DoubleConsumer consumer;
 
    public ForEachDoubleOperation(Iterable<IntermediateOperation> intermediateOperations,
            Supplier<Stream<CacheEntry>> supplier, int batchSize, DoubleConsumer consumer) {
-      super(intermediateOperations, supplier);
-      this.batchSize = batchSize;
+      super(intermediateOperations, supplier, batchSize);
       this.consumer = consumer;
    }
 
    @Override
-   public boolean lostSegment(boolean stopIfLost) {
-      // TODO: stop this early
-      return true;
-   }
-
-   @Override
-   public List<Double> performOperation(IntermediateCollector<Collection<Double>> response) {
-      /**
-       * This is for rehash only! {@link org.infinispan.stream.impl.termop.SingleRunOperation} should always be used for
-       * non rehash
-       */
-      throw new UnsupportedOperationException();
-   }
-
-   @Override
-   public Collection<CacheEntry<K, K>> performOperationRehashAware(
-           IntermediateCollector<Collection<CacheEntry<K, K>>> response) {
-      // We only support sequential streams for iterator rehash aware
-      BaseStream<?, ?> stream = supplier.get().sequential();
-
-      List<CacheEntry<K, K>> collectedValues = new ArrayList(batchSize);
-
-      double[] list = new double[batchSize];
-      AtomicInteger offset = new AtomicInteger();
-      Object[] currentKey = new Object[1];
-      stream = ((Stream<Map.Entry<K, ?>>) stream).peek(e -> {
-         if (offset.get() > 0) {
-            collectedValues.add(new ImmortalCacheEntry(currentKey[0], currentKey[0]));
-            if (collectedValues.size() >= batchSize) {
-               for (int i = 0; i < offset.get(); ++i) {
-                  consumer.accept(list[i]);
-               }
-               response.sendDataResonse(collectedValues);
-               collectedValues.clear();
-               offset.set(0);
-            }
-         }
-         currentKey[0] = e.getKey();
-      });
-      for (IntermediateOperation intermediateOperation : intermediateOperations) {
-         stream = intermediateOperation.perform(stream);
+   protected void handleArray(double[] array, int size) {
+      for (int i = 0; i < size; ++i) {
+         consumer.accept(array[i]);
       }
-
-      DoubleStream convertedStream = ((DoubleStream)stream);
-      // We rely on the fact that iterator processes 1 entry at a time when sequential
-      convertedStream.forEach(d -> list[offset.getAndIncrement()] = d);
-      if (offset.get() > 0) {
-         for (int i = 0; i < offset.get(); ++i) {
-            consumer.accept(list[i]);
-         }
-         collectedValues.add(new ImmortalCacheEntry(currentKey[0], currentKey[0]));
-      }
-      return collectedValues;
-   }
-
-   public int getBatchSize() {
-      return batchSize;
    }
 
    public DoubleConsumer getConsumer() {
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachFlatMapObjDoubleOperation.java b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachFlatMapObjDoubleOperation.java
new file mode 100644
index 000000000000..c9e59a5f9447
--- /dev/null
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachFlatMapObjDoubleOperation.java
@@ -0,0 +1,51 @@
+package org.infinispan.stream.impl.termop.primitive;
+
+import org.infinispan.Cache;
+import org.infinispan.container.entries.CacheEntry;
+import org.infinispan.factories.ComponentRegistry;
+import org.infinispan.stream.CacheAware;
+import org.infinispan.stream.impl.intops.IntermediateOperation;
+import org.infinispan.stream.impl.termop.AbstractForEachOperation;
+
+import java.util.List;
+import java.util.function.DoubleConsumer;
+import java.util.function.ObjDoubleConsumer;
+import java.util.function.Supplier;
+import java.util.stream.DoubleStream;
+import java.util.stream.Stream;
+
+/**
+ * Terminal rehash aware operation that handles for each where flat map operation is performed on a
+ * {@link DoubleStream}.
+ * @param <K> key type of the supplied stream
+ */
+public class ForEachFlatMapObjDoubleOperation<K> extends AbstractForEachOperation<K, Double, DoubleStream> {
+   private final ObjDoubleConsumer<Cache<K, ?>> consumer;
+   private transient Cache<K, ?> cache;
+
+   public ForEachFlatMapObjDoubleOperation(Iterable<IntermediateOperation> intermediateOperations,
+           Supplier<Stream<CacheEntry>> supplier, int batchSize, ObjDoubleConsumer<Cache<K, ?>> consumer) {
+      super(intermediateOperations, supplier, batchSize);
+      this.consumer = consumer;
+   }
+
+   @Override
+   protected void handleList(List<Double> list) {
+      list.forEach(d -> consumer.accept(cache, d));
+   }
+
+   @Override
+   protected void handleStreamForEach(DoubleStream stream, List<Double> list) {
+      stream.forEach(list::add);
+   }
+
+   public ObjDoubleConsumer<Cache<K, ?>> getConsumer() {
+      return consumer;
+   }
+
+   @Override
+   public void handleInjection(ComponentRegistry registry) {
+      super.handleInjection(registry);
+      cache = registry.getComponent(Cache.class);
+   }
+}
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachFlatMapObjIntOperation.java b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachFlatMapObjIntOperation.java
new file mode 100644
index 000000000000..71128c7b8498
--- /dev/null
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachFlatMapObjIntOperation.java
@@ -0,0 +1,49 @@
+package org.infinispan.stream.impl.termop.primitive;
+
+import org.infinispan.Cache;
+import org.infinispan.container.entries.CacheEntry;
+import org.infinispan.factories.ComponentRegistry;
+import org.infinispan.stream.impl.intops.IntermediateOperation;
+import org.infinispan.stream.impl.termop.AbstractForEachOperation;
+
+import java.util.List;
+import java.util.function.ObjIntConsumer;
+import java.util.function.Supplier;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+/**
+ * Terminal rehash aware operation that handles for each where flat map operation is performed on a
+ * {@link IntStream}.
+ * @param <K> key type of the supplied stream
+ */
+public class ForEachFlatMapObjIntOperation<K> extends AbstractForEachOperation<K, Integer, IntStream> {
+   private final ObjIntConsumer<Cache<K, ?>> consumer;
+   private transient Cache<K, ?> cache;
+
+   public ForEachFlatMapObjIntOperation(Iterable<IntermediateOperation> intermediateOperations,
+           Supplier<Stream<CacheEntry>> supplier, int batchSize, ObjIntConsumer<Cache<K, ?>> consumer) {
+      super(intermediateOperations, supplier, batchSize);
+      this.consumer = consumer;
+   }
+
+   @Override
+   protected void handleList(List<Integer> list) {
+      list.forEach(d -> consumer.accept(cache, d));
+   }
+
+   @Override
+   protected void handleStreamForEach(IntStream stream, List<Integer> list) {
+      stream.forEach(list::add);
+   }
+
+   public ObjIntConsumer<Cache<K, ?>> getConsumer() {
+      return consumer;
+   }
+
+   @Override
+   public void handleInjection(ComponentRegistry registry) {
+      super.handleInjection(registry);
+      cache = registry.getComponent(Cache.class);
+   }
+}
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachFlatMapObjLongOperation.java b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachFlatMapObjLongOperation.java
new file mode 100644
index 000000000000..742101e03723
--- /dev/null
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachFlatMapObjLongOperation.java
@@ -0,0 +1,49 @@
+package org.infinispan.stream.impl.termop.primitive;
+
+import org.infinispan.Cache;
+import org.infinispan.container.entries.CacheEntry;
+import org.infinispan.factories.ComponentRegistry;
+import org.infinispan.stream.impl.intops.IntermediateOperation;
+import org.infinispan.stream.impl.termop.AbstractForEachOperation;
+
+import java.util.List;
+import java.util.function.ObjLongConsumer;
+import java.util.function.Supplier;
+import java.util.stream.LongStream;
+import java.util.stream.Stream;
+
+/**
+ * Terminal rehash aware operation that handles for each where flat map operation is performed on a
+ * {@link LongStream}.
+ * @param <K> key type of the supplied stream
+ */
+public class ForEachFlatMapObjLongOperation<K> extends AbstractForEachOperation<K, Long, LongStream> {
+   private final ObjLongConsumer<Cache<K, ?>> consumer;
+   private transient Cache<K, ?> cache;
+
+   public ForEachFlatMapObjLongOperation(Iterable<IntermediateOperation> intermediateOperations,
+           Supplier<Stream<CacheEntry>> supplier, int batchSize, ObjLongConsumer<Cache<K, ?>> consumer) {
+      super(intermediateOperations, supplier, batchSize);
+      this.consumer = consumer;
+   }
+
+   @Override
+   protected void handleList(List<Long> list) {
+      list.forEach(d -> consumer.accept(cache, d));
+   }
+
+   @Override
+   protected void handleStreamForEach(LongStream stream, List<Long> list) {
+      stream.forEach(list::add);
+   }
+
+   public ObjLongConsumer<Cache<K, ?>> getConsumer() {
+      return consumer;
+   }
+
+   @Override
+   public void handleInjection(ComponentRegistry registry) {
+      super.handleInjection(registry);
+      cache = registry.getComponent(Cache.class);
+   }
+}
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachIntOperation.java b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachIntOperation.java
index d1fc35ac0aec..41689e789184 100644
--- a/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachIntOperation.java
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachIntOperation.java
@@ -2,22 +2,12 @@
 
 import org.infinispan.Cache;
 import org.infinispan.container.entries.CacheEntry;
-import org.infinispan.container.entries.ImmortalCacheEntry;
 import org.infinispan.factories.ComponentRegistry;
-import org.infinispan.stream.impl.KeyTrackingTerminalOperation;
-import org.infinispan.stream.impl.intops.IntermediateOperation;
-import org.infinispan.stream.impl.termop.BaseTerminalOperation;
-import org.infinispan.stream.impl.termop.object.NoMapIteratorOperation;
 import org.infinispan.stream.CacheAware;
+import org.infinispan.stream.impl.intops.IntermediateOperation;
 
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.IntConsumer;
 import java.util.function.Supplier;
-import java.util.stream.BaseStream;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
@@ -26,74 +16,20 @@
  * {@link IntStream}. Note this means it is an implied map intermediate operation.
  * @param <K> key type of the supplied stream
  */
-public class ForEachIntOperation<K> extends BaseTerminalOperation implements KeyTrackingTerminalOperation<K, Integer, K> {
-   private final int batchSize;
+public class ForEachIntOperation<K> extends AbstractForEachIntOperation<K> {
    private final IntConsumer consumer;
 
    public ForEachIntOperation(Iterable<IntermediateOperation> intermediateOperations,
            Supplier<Stream<CacheEntry>> supplier, int batchSize, IntConsumer consumer) {
-      super(intermediateOperations, supplier);
-      this.batchSize = batchSize;
+      super(intermediateOperations, supplier, batchSize);
       this.consumer = consumer;
    }
 
    @Override
-   public boolean lostSegment(boolean stopIfLost) {
-      // TODO: stop this early
-      return true;
-   }
-
-   @Override
-   public List<Integer> performOperation(IntermediateCollector<Collection<Integer>> response) {
-      /**
-       * This is for rehash only! {@link NoMapIteratorOperation} should always be used for non rehash
-       */
-      throw new UnsupportedOperationException();
-   }
-
-   @Override
-   public Collection<CacheEntry<K, K>> performOperationRehashAware(
-           IntermediateCollector<Collection<CacheEntry<K, K>>> response) {
-      // We only support sequential streams for iterator rehash aware
-      BaseStream<?, ?> stream = supplier.get().sequential();
-
-      List<CacheEntry<K, K>> collectedValues = new ArrayList(batchSize);
-
-      int[] list = new int[batchSize];
-      AtomicInteger offset = new AtomicInteger();
-      Object[] currentKey = new Object[1];
-      stream = ((Stream<Map.Entry<K, ?>>) stream).peek(e -> {
-         if (offset.get() > 0) {
-            collectedValues.add(new ImmortalCacheEntry(currentKey[0], currentKey[0]));
-            if (collectedValues.size() >= batchSize) {
-               for (int i = 0; i < offset.get(); ++i) {
-                  consumer.accept(list[i]);
-               }
-               response.sendDataResonse(collectedValues);
-               collectedValues.clear();
-               offset.set(0);
-            }
-         }
-         currentKey[0] = e.getKey();
-      });
-      for (IntermediateOperation intermediateOperation : intermediateOperations) {
-         stream = intermediateOperation.perform(stream);
+   protected void handleArray(int[] array, int size) {
+      for (int i = 0; i < size; ++i) {
+         consumer.accept(array[i]);
       }
-
-      IntStream convertedStream = ((IntStream)stream);
-      // We rely on the fact that iterator processes 1 entry at a time when sequential
-      convertedStream.forEach(i -> list[offset.getAndIncrement()] = i);
-      if (offset.get() > 0) {
-         for (int i = 0; i < offset.get(); ++i) {
-            consumer.accept(list[i]);
-         }
-         collectedValues.add(new ImmortalCacheEntry(currentKey[0], currentKey[0]));
-      }
-      return collectedValues;
-   }
-
-   public int getBatchSize() {
-      return batchSize;
    }
 
    public IntConsumer getConsumer() {
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachLongOperation.java b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachLongOperation.java
index 2ddb7d896e91..d068748c1eb1 100644
--- a/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachLongOperation.java
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachLongOperation.java
@@ -2,22 +2,12 @@
 
 import org.infinispan.Cache;
 import org.infinispan.container.entries.CacheEntry;
-import org.infinispan.container.entries.ImmortalCacheEntry;
 import org.infinispan.factories.ComponentRegistry;
-import org.infinispan.stream.impl.KeyTrackingTerminalOperation;
-import org.infinispan.stream.impl.intops.IntermediateOperation;
-import org.infinispan.stream.impl.termop.BaseTerminalOperation;
-import org.infinispan.stream.impl.termop.object.NoMapIteratorOperation;
 import org.infinispan.stream.CacheAware;
+import org.infinispan.stream.impl.intops.IntermediateOperation;
 
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.LongConsumer;
 import java.util.function.Supplier;
-import java.util.stream.BaseStream;
 import java.util.stream.LongStream;
 import java.util.stream.Stream;
 
@@ -26,74 +16,20 @@
  * {@link LongStream}. Note this means it is an implied map intermediate operation.
  * @param <K> key type of the supplied stream
  */
-public class ForEachLongOperation<K> extends BaseTerminalOperation implements KeyTrackingTerminalOperation<K, Long, K> {
-   private final int batchSize;
+public class ForEachLongOperation<K> extends AbstractForEachLongOperation<K> {
    private final LongConsumer consumer;
 
    public ForEachLongOperation(Iterable<IntermediateOperation> intermediateOperations,
            Supplier<Stream<CacheEntry>> supplier, int batchSize, LongConsumer consumer) {
-      super(intermediateOperations, supplier);
-      this.batchSize = batchSize;
+      super(intermediateOperations, supplier, batchSize);
       this.consumer = consumer;
    }
 
    @Override
-   public boolean lostSegment(boolean stopIfLost) {
-      // TODO: stop this early
-      return true;
-   }
-
-   @Override
-   public List<Long> performOperation(IntermediateCollector<Collection<Long>> response) {
-      /**
-       * This is for rehash only! {@link NoMapIteratorOperation} should always be used for non rehash
-       */
-      throw new UnsupportedOperationException();
-   }
-
-   @Override
-   public Collection<CacheEntry<K, K>> performOperationRehashAware(
-           IntermediateCollector<Collection<CacheEntry<K, K>>> response) {
-      // We only support sequential streams for iterator rehash aware
-      BaseStream<?, ?> stream = supplier.get().sequential();
-
-      List<CacheEntry<K, K>> collectedValues = new ArrayList(batchSize);
-
-      long[] list = new long[batchSize];
-      AtomicInteger offset = new AtomicInteger();
-      Object[] currentKey = new Object[1];
-      stream = ((Stream<Map.Entry<K, ?>>) stream).peek(e -> {
-         if (offset.get() > 0) {
-            collectedValues.add(new ImmortalCacheEntry(currentKey[0], currentKey[0]));
-            if (collectedValues.size() >= batchSize) {
-               for (int i = 0; i < offset.get(); ++i) {
-                  consumer.accept(list[i]);
-               }
-               response.sendDataResonse(collectedValues);
-               collectedValues.clear();
-               offset.set(0);
-            }
-         }
-         currentKey[0] = e.getKey();
-      });
-      for (IntermediateOperation intermediateOperation : intermediateOperations) {
-         stream = intermediateOperation.perform(stream);
+   protected void handleArray(long[] array, int size) {
+      for (int i = 0; i < size; i++) {
+         consumer.accept(array[i]);
       }
-
-      LongStream convertedStream = ((LongStream)stream);
-      // We rely on the fact that iterator processes 1 entry at a time when sequential
-      convertedStream.forEach(d -> list[offset.getAndIncrement()] = d);
-      if (offset.get() > 0) {
-         for (int i = 0; i < offset.get(); ++i) {
-            consumer.accept(list[i]);
-         }
-         collectedValues.add(new ImmortalCacheEntry(currentKey[0], currentKey[0]));
-      }
-      return collectedValues;
-   }
-
-   public int getBatchSize() {
-      return batchSize;
    }
 
    public LongConsumer getConsumer() {
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachObjDoubleOperation.java b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachObjDoubleOperation.java
new file mode 100644
index 000000000000..86002d6ec5c4
--- /dev/null
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachObjDoubleOperation.java
@@ -0,0 +1,44 @@
+package org.infinispan.stream.impl.termop.primitive;
+
+import org.infinispan.Cache;
+import org.infinispan.container.entries.CacheEntry;
+import org.infinispan.factories.ComponentRegistry;
+import org.infinispan.stream.impl.intops.IntermediateOperation;
+
+import java.util.function.ObjDoubleConsumer;
+import java.util.function.Supplier;
+import java.util.stream.DoubleStream;
+import java.util.stream.Stream;
+
+/**
+ * Terminal rehash aware operation that handles for each where no flat map operations are defined on a
+ * {@link DoubleStream}. Note this means it is an implied map intermediate operation.
+ * @param <K> key type of the supplied stream
+ */
+public class ForEachObjDoubleOperation<K> extends AbstractForEachDoubleOperation<K> {
+   private final ObjDoubleConsumer<Cache<K, ?>> consumer;
+   private transient Cache<K, ?> cache;
+
+   public ForEachObjDoubleOperation(Iterable<IntermediateOperation> intermediateOperations,
+           Supplier<Stream<CacheEntry>> supplier, int batchSize, ObjDoubleConsumer<Cache<K, ?>> consumer) {
+      super(intermediateOperations, supplier, batchSize);
+      this.consumer = consumer;
+   }
+
+   @Override
+   protected void handleArray(double[] array, int size) {
+      for (int i = 0; i < size; ++i) {
+         consumer.accept(cache, array[i]);
+      }
+   }
+
+   public ObjDoubleConsumer<Cache<K, ?>> getConsumer() {
+      return consumer;
+   }
+
+   @Override
+   public void handleInjection(ComponentRegistry registry) {
+      super.handleInjection(registry);
+      cache = registry.getComponent(Cache.class);
+   }
+}
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachObjIntOperation.java b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachObjIntOperation.java
new file mode 100644
index 000000000000..bb73ff5f8097
--- /dev/null
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachObjIntOperation.java
@@ -0,0 +1,44 @@
+package org.infinispan.stream.impl.termop.primitive;
+
+import org.infinispan.Cache;
+import org.infinispan.container.entries.CacheEntry;
+import org.infinispan.factories.ComponentRegistry;
+import org.infinispan.stream.impl.intops.IntermediateOperation;
+
+import java.util.function.ObjIntConsumer;
+import java.util.function.Supplier;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+/**
+ * Terminal rehash aware operation that handles for each where no flat map operations are defined on a
+ * {@link IntStream}. Note this means it is an implied map intermediate operation.
+ * @param <K> key type of the supplied stream
+ */
+public class ForEachObjIntOperation<K> extends AbstractForEachIntOperation<K> {
+   private final ObjIntConsumer<Cache<K, ?>> consumer;
+   private transient Cache<K, ?> cache;
+
+   public ForEachObjIntOperation(Iterable<IntermediateOperation> intermediateOperations,
+           Supplier<Stream<CacheEntry>> supplier, int batchSize, ObjIntConsumer<Cache<K, ?>> consumer) {
+      super(intermediateOperations, supplier, batchSize);
+      this.consumer = consumer;
+   }
+
+   @Override
+   protected void handleArray(int[] array, int size) {
+      for (int i = 0; i < size; ++i) {
+         consumer.accept(cache, array[i]);
+      }
+   }
+
+   public ObjIntConsumer<Cache<K, ?>> getConsumer() {
+      return consumer;
+   }
+
+   @Override
+   public void handleInjection(ComponentRegistry registry) {
+      super.handleInjection(registry);
+      cache = registry.getComponent(Cache.class);
+   }
+}
diff --git a/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachObjLongOperation.java b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachObjLongOperation.java
new file mode 100644
index 000000000000..ab90ad52719e
--- /dev/null
+++ b/core/src/main/java/org/infinispan/stream/impl/termop/primitive/ForEachObjLongOperation.java
@@ -0,0 +1,44 @@
+package org.infinispan.stream.impl.termop.primitive;
+
+import org.infinispan.Cache;
+import org.infinispan.container.entries.CacheEntry;
+import org.infinispan.factories.ComponentRegistry;
+import org.infinispan.stream.impl.intops.IntermediateOperation;
+
+import java.util.function.ObjLongConsumer;
+import java.util.function.Supplier;
+import java.util.stream.LongStream;
+import java.util.stream.Stream;
+
+/**
+ * Terminal rehash aware operation that handles for each where no flat map operations are defined on a
+ * {@link LongStream}. Note this means it is an implied map intermediate operation.
+ * @param <K> key type of the supplied stream
+ */
+public class ForEachObjLongOperation<K> extends AbstractForEachLongOperation<K> {
+   private final ObjLongConsumer<Cache<K, ?>> consumer;
+   private transient Cache<K, ?> cache;
+
+   public ForEachObjLongOperation(Iterable<IntermediateOperation> intermediateOperations,
+           Supplier<Stream<CacheEntry>> supplier, int batchSize, ObjLongConsumer<Cache<K, ?>> consumer) {
+      super(intermediateOperations, supplier, batchSize);
+      this.consumer = consumer;
+   }
+
+   @Override
+   protected void handleArray(long[] array, int size) {
+      for (int i = 0; i < size; ++i) {
+         consumer.accept(cache, array[i]);
+      }
+   }
+
+   public ObjLongConsumer<Cache<K, ?>> getConsumer() {
+      return consumer;
+   }
+
+   @Override
+   public void handleInjection(ComponentRegistry registry) {
+      super.handleInjection(registry);
+      cache = registry.getComponent(Cache.class);
+   }
+}
diff --git a/core/src/test/java/org/infinispan/stream/BaseStreamTest.java b/core/src/test/java/org/infinispan/stream/BaseStreamTest.java
index 7cd219cd7035..ca06ad189703 100644
--- a/core/src/test/java/org/infinispan/stream/BaseStreamTest.java
+++ b/core/src/test/java/org/infinispan/stream/BaseStreamTest.java
@@ -317,6 +317,29 @@ public void testObjForEachCacheInjected() {
       }
    }
 
+   public void testObjForEachBiConsumer() {
+      Cache<Integer, String> cache = getCache(0);
+
+      int cacheOffset = populateNextForEachStructure(cache);
+      int atomicOffset = populateNextForEachStructure(new AtomicInteger());
+      try {
+         testIntOperation(() -> {
+            createStream(cache.entrySet()).forEach((c, e) -> {
+               Cache<?, ?> localCache = getForEachObject(cacheOffset);
+               if (c != null && localCache != null && c.getName().equals(localCache.getName())) {
+                  ((AtomicInteger) getForEachObject(atomicOffset)).addAndGet(e.getKey());
+               } else {
+                  fail("Did not receive correct cache!");
+               }
+            });
+            return ((AtomicInteger) getForEachObject(atomicOffset)).get();
+         }, cache);
+      } finally {
+         clearForEachObject(cacheOffset);
+         clearForEachObject(atomicOffset);
+      }
+   }
+
    public void testObjKeySetForEachCacheInjected() {
       Cache<Integer, String> cache = getCache(0);
 
@@ -808,6 +831,53 @@ public void testIntForEachCacheInjected() {
       }
    }
 
+   public void testIntForEachBiConsumer() {
+      Cache<Integer, String> cache = getCache(0);
+      int cacheOffset = populateNextForEachStructure(cache);
+      int atomicOffset = populateNextForEachStructure(new AtomicInteger());
+
+      try {
+         testIntOperation(() -> {
+            createStream(cache.entrySet()).mapToInt(toInt).forEach((c, i) -> {
+               Cache<?, ?> localCache = getForEachObject(cacheOffset);
+               if (c != null && localCache != null && c.getName().equals(localCache.getName())) {
+                  AtomicInteger atomicInteger = getForEachObject(atomicOffset);
+                  atomicInteger.addAndGet(i);
+               }
+            });
+            return ((AtomicInteger) getForEachObject(atomicOffset)).get();
+         }, cache);
+      } finally {
+         clearForEachObject(cacheOffset);
+         clearForEachObject(atomicOffset);
+      }
+   }
+
+   public void testIntFlatMapObjConsumerForEach() {
+      Cache<Integer, String> cache = getCache(0);
+      String cacheName = cache.getName();
+      int range = 10;
+      // First populate the cache with a bunch of values
+      IntStream.range(0, range).boxed().forEach(i -> cache.put(i, i + "-value"));
+
+      assertEquals(range, cache.size());
+      CacheSet<Map.Entry<Integer, String>> entrySet = cache.entrySet();
+
+      int offset = populateNextForEachStructure(new AtomicInteger());
+      try {
+         createStream(entrySet).distributedBatchSize(5).mapToInt(toInt).flatMap(i -> IntStream.of(i, 2))
+                 .forEach((c, e) -> {
+                    assertEquals(cacheName, c.getName());
+                    AtomicInteger atomic = getForEachObject(offset);
+                    atomic.addAndGet(e);
+                 });
+         AtomicInteger atomic = getForEachObject(offset);
+         assertEquals((range - 1) * (range / 2) + 2 * range, atomic.get());
+      } finally {
+         clearForEachObject(offset);
+      }
+   }
+
    public void testIntIterator() {
       Cache<Integer, String> cache = getCache(0);
 
@@ -1186,6 +1256,53 @@ public void testLongForEachCacheInjected() {
       }
    }
 
+   public void testLongForEachBiConsumer() {
+      Cache<Long, String> cache = getCache(0);
+      int cacheOffset = populateNextForEachStructure(cache);
+      int atomicOffset = populateNextForEachStructure(new AtomicLong());
+
+      try {
+         testLongOperation(() -> {
+            createStream(cache.entrySet()).mapToLong(toLong).forEach((c, i) -> {
+               Cache<?, ?> localCache = getForEachObject(cacheOffset);
+               if (c != null && localCache != null && c.getName().equals(localCache.getName())) {
+                  AtomicLong atomicLong = getForEachObject(atomicOffset);
+                  atomicLong.addAndGet(i);
+               }
+            });
+            return ((AtomicLong) getForEachObject(atomicOffset)).get();
+         }, cache);
+      } finally {
+         clearForEachObject(cacheOffset);
+         clearForEachObject(atomicOffset);
+      }
+   }
+
+   public void testLongFlatMapObjConsumerForEach() {
+      Cache<Long, String> cache = getCache(0);
+      String cacheName = cache.getName();
+      int range = 10;
+      // First populate the cache with a bunch of values
+      LongStream.range(0, range).boxed().forEach(i -> cache.put(i, i + "-value"));
+
+      assertEquals(range, cache.size());
+      CacheSet<Map.Entry<Long, String>> entrySet = cache.entrySet();
+
+      int offset = populateNextForEachStructure(new AtomicLong());
+      try {
+         createStream(entrySet).distributedBatchSize(5).mapToLong(toLong).flatMap(i -> LongStream.of(i, 2))
+                 .forEach((c, e) -> {
+                    assertEquals(cacheName, c.getName());
+                    AtomicLong atomic = getForEachObject(offset);
+                    atomic.addAndGet(e);
+                 });
+         AtomicLong atomic = getForEachObject(offset);
+         assertEquals((range - 1) * (range / 2) + 2 * range, atomic.get());
+      } finally {
+         clearForEachObject(offset);
+      }
+   }
+
    public void testLongIterator() {
       Cache<Long, String> cache = getCache(0);
 
@@ -1582,6 +1699,62 @@ public void testDoubleForEachCacheInjected() {
       }
    }
 
+   public void testDoubleForEachBiConsumer() {
+      Cache<Double, String> cache = getCache(0);
+      int cacheOffset = populateNextForEachStructure(cache);
+      int offset = populateNextForEachStructure(new DoubleSummaryStatistics());
+
+      try {
+         testDoubleOperation(() -> {
+            createStream(cache.entrySet()).mapToDouble(toDouble).forEach((c, d) -> {
+               Cache<?, ?> localCache = getForEachObject(cacheOffset);
+               if (c != null && localCache != null && c.getName().equals(localCache.getName())) {
+                  DoubleSummaryStatistics stats = getForEachObject(offset);
+                  synchronized (stats) {
+                     stats.accept(d);
+                  }
+               }
+            });
+            DoubleSummaryStatistics stats = getForEachObject(offset);
+            return stats;
+         }, cache);
+      } finally {
+         clearForEachObject(cacheOffset);
+         clearForEachObject(offset);
+      }
+   }
+
+   public void testDoubleFlatMapObjConsumerForEach() {
+      Cache<Double, String> cache = getCache(0);
+      String cacheName = cache.getName();
+      int range = 10;
+      // First populate the cache with a bunch of values
+      DoubleStream.iterate(0.0, d -> d + .5).limit(10).boxed().forEach(i -> cache.put(i, i + "-value"));
+
+      assertEquals(range, cache.size());
+      CacheSet<Map.Entry<Double, String>> entrySet = cache.entrySet();
+
+      int offset = populateNextForEachStructure(new DoubleSummaryStatistics());
+      try {
+         createStream(entrySet).distributedBatchSize(5).mapToDouble(toDouble).flatMap(e -> DoubleStream.of(e, 2.25))
+                 .forEach((c, e) -> {
+                    assertEquals(cacheName, c.getName());
+                    DoubleSummaryStatistics stats = getForEachObject(offset);
+                    synchronized (stats) {
+                       stats.accept(e);
+                    }
+                 });
+         DoubleSummaryStatistics stats = getForEachObject(offset);
+         assertEquals(2.25, stats.getAverage());
+         assertEquals(0.0, stats.getMin());
+         assertEquals(4.5, stats.getMax());
+         assertEquals(20, stats.getCount());
+         assertEquals(45.0, stats.getSum());
+      } finally {
+         clearForEachObject(offset);
+      }
+   }
+
    public void testDoubleIterator() {
       Cache<Double, String> cache = getCache(0);
       testDoubleOperation(() -> {