Distributed Stream rehash is broken if a map operation is used

Rehash distributed streams are broken if an intermediate operation is present that has a map operation.

cache.entrySet().stream().map(Map.Entry::getValue)

This is due to the stream supplier doesn't create a new stream on each invocation in DistributionBulkInterceptor.