Due to the nature of how streams work with retries and performing the operation on the primary owner, this works out quite well with forEach to be done in an efficient way.

The problem is that this only really works well with non tx and pessimistic tx. This obviously leaves out optimistic tx, which at first I was a little worried about. But after thinking about it more, this prelocking and optimistic tx don't really fit that well together anyways. So I am thinking whenever this operation is performed it would throw an exception not letting the user use this feature in optimistic transactions.

Another question is what does the API for this look like. I was debating between 3 options myself:

1. AdvancedCache.forEachWithLock(BiConsumer<Cache, CacheEntry<K, V>> consumer)

This require the least amount of changes, however the user can't customize certain parameters that CacheStream currently provides (listed below - big one being filterKeys).

2. CacheStream.forEachWithLock(BiConsumer<Cache, CacheEntry<K, V>> consumer)

 
This method would only be allowed to be invoked on the Stream if no other intermediate operations were invoked, otherwise an exception would be thrown. This still gives us access to all of the CacheStream methods that aren't on the Stream interface (ie. sequentialDistribution, parallelDistribution, parallel, sequential, filterKeys, filterKeySegments, distributedBatchSize, disableRehashAware, timeout).

3. LockedStream<CacheEntry<K, V>> AdvancedCache.lockedStream()

This requires the most changes, however the API would be the most explicit. In this case the LockedStream would only have the methods on it that are able to be invoked as noted above and forEach.

I personally feel that #3 might be the cleanest, but obviously requires adding more classes. Let me know what you guys think and if you think the optimistic exclusion is acceptable.

Thanks,

 - Will