2.x: fix cross-boundary invalid fusion with observeOn, flatMap & zip

[akarnokd]

When flatMap and zip fuses its sources, it was possible one of the async source polls on another source which executed boundary-sensitive operators (map, filter) on the wrong thread.

For clarity, here is a diagram showing the execution flow of a classical and fused setup:

In the classical flow, everything is push and when flatMap collects the available elements, all side-effects happened inside map.

In the fused flow, there are no queues and the onNext call is an indication to poll() on the sources inside flatMap (or zip). If the first source triggers onNext, that source is correctly polled and map executes on the right thread. However, when the flatMap continues to collect other available elements, it polls on the other source and executes that map still on the first scheduler, despite that source having its own scheduler specified.

The solution is to mark flatMap and zip's inner consumer as boundary sensitive which prevents the fusion above since map is also marked as boundary sensitive.

Related: reactor/reactor-core#342

[codecov-io]

Current coverage is 95.55% (diff: 100%)

Merging #4984 into 2.x will decrease coverage by 0.07%

@@                2.x      #4984   diff @@
==========================================
  Files           592        592          
  Lines         37968      37968          
  Methods           0          0          
  Messages          0          0          
  Branches       5752       5752          
==========================================
- Hits          36307      36279    -28   
- Misses          701        735    +34   
+ Partials        960        954     -6   

Powered by Codecov. Last update 5717827...0664155