2.x: fix PublishProcessor cancel/emission overflow bug

[akarnokd]

This PR should fix the bug that caused the test failure in #5545.

The bug manifested itself when a cancellation was happening the same time a request 1 was being fulfilled. Since the same request accounting was used for cancellation indicator, if the cancel happened between the onNext()'s get() check and decrementAndGet, this decrementAndGet decremented Long.MIN_VALUE unconditionally, which lead to a state that would appear the Subscriber still can receive events. A concurrent offer, which saves the current array of registered Subscribers, then would emit an item and overflow the Subscriber.

The fix is to use the cancellation-aware BackpressureHelper.producedCancel() utility.

Unit test were added to verify the correct behavior on both PublishProcessor and BehaviorProcessor (the latter uses different cancellation mechanism via a dedicated field).

[codecov]

Codecov Report

Merging #5669 into 2.x will decrease coverage by 0.05%.
The diff coverage is 100%.

@@             Coverage Diff              @@
##                2.x    #5669      +/-   ##
============================================
- Coverage     96.27%   96.21%   -0.06%     
- Complexity     5815     5818       +3     
============================================
  Files           633      633              
  Lines         41553    41552       -1     
  Branches       5752     5751       -1     
============================================
- Hits          40004    39981      -23     
- Misses          612      625      +13     
- Partials        937      946       +9

Impacted Files	Coverage Δ	Complexity Δ
...java/io/reactivex/processors/PublishProcessor.java	98.27% <100%> (-0.02%)	45 <0> (ø)
...l/operators/observable/ObservableFlatMapMaybe.java	84.96% <0%> (-8.5%)	2% <0%> (ø)
...ernal/operators/flowable/FlowableFlatMapMaybe.java	88.88% <0%> (-5.8%)	2% <0%> (ø)
...nternal/operators/observable/ObservableCreate.java	94.01% <0%> (-4.28%)	2% <0%> (ø)
...rnal/operators/flowable/FlowableTakeLastTimed.java	96.29% <0%> (-2.78%)	2% <0%> (ø)
...nternal/operators/parallel/ParallelSortedJoin.java	92.75% <0%> (-2.18%)	2% <0%> (ø)
...ernal/operators/flowable/FlowableFromIterable.java	95.18% <0%> (-1.07%)	5% <0%> (ø)
.../operators/maybe/MaybeFlatMapIterableFlowable.java	97.54% <0%> (-0.82%)	2% <0%> (ø)
...ernal/operators/flowable/FlowableTimeoutTimed.java	98.37% <0%> (-0.82%)	3% <0%> (ø)
.../operators/observable/ObservableFlatMapSingle.java	88.8% <0%> (-0.75%)	2% <0%> (ø)
... and 19 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update cd6bc08...678c62f. Read the comment docs.