3.x: Fix Flowable.groupBy eviction-completion-replenishment problems

[akarnokd]

This PR fixes the two cases from #6982 (comment)

Case 1: completions missed

When the upstream completed and some groups were waiting in the eviction queue, the eviction queue was cleared without completing those groups. Since they were no longer in the main map, those groups never completed and downstream operators relying on all groups completing (i.e., flatMap) would never complete.

The fix is to call completeEvictions upon upstream termination.

Case 2: no replenisment on evicted groups

When an eviction happens, the groups get completed. However, unlike with the non-evicting case, a group completion no longer implies the upstream has completed. Consequently, the evicted groups emitted their items but never issued a replenishment for them, hanging the still-alive main operator.

The fix is to replenish emitted (and polled) counts whenever the group detects a completion. (Errors still imply the main operator has terminated so no need to replenish then.)

Resolves #6982

[akarnokd]

/cc @davidmoten @dbakr

[codecov]

Codecov Report

Merging #6988 into 3.x will decrease coverage by 0.04%.
The diff coverage is 100.00%.

@@             Coverage Diff              @@
##                3.x    #6988      +/-   ##
============================================
- Coverage     99.53%   99.48%   -0.05%     
+ Complexity     6670     6668       -2     
============================================
  Files           742      742              
  Lines         47267    47269       +2     
  Branches       6374     6372       -2     
============================================
- Hits          47045    47027      -18     
- Misses           99      111      +12     
- Partials        123      131       +8     

Impacted Files	Coverage Δ	Complexity Δ
...3/internal/operators/flowable/FlowableGroupBy.java	85.02% <100.00%> (+0.09%)	3.00 <0.00> (ø)
...tivex/rxjava3/internal/jdk8/ParallelCollector.java	91.74% <0.00%> (-6.43%)	2.00% <0.00%> (ø%)
.../operators/flowable/FlowableBlockingSubscribe.java	93.02% <0.00%> (-4.66%)	10.00% <0.00%> (-1.00%)
...rnal/operators/flowable/FlowableFlatMapSingle.java	94.18% <0.00%> (-3.49%)	2.00% <0.00%> (ø%)
...ernal/operators/flowable/FlowableFlatMapMaybe.java	95.85% <0.00%> (-2.08%)	2.00% <0.00%> (ø%)
...nternal/operators/observable/ObservableCreate.java	97.43% <0.00%> (-1.71%)	2.00% <0.00%> (ø%)
...ternal/operators/observable/ObservableGroupBy.java	98.90% <0.00%> (-1.10%)	2.00% <0.00%> (ø%)
.../operators/observable/ObservableFlatMapSingle.java	94.44% <0.00%> (-0.80%)	2.00% <0.00%> (ø%)
...va3/internal/operators/flowable/FlowableCache.java	99.24% <0.00%> (-0.76%)	38.00% <0.00%> (-1.00%)
.../reactivex/rxjava3/processors/ReplayProcessor.java	99.19% <0.00%> (-0.41%)	52.00% <0.00%> (ø%)
... and 6 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 125bd05...687ac53. Read the comment docs.

[davidmoten]

thanks @akarnokd

[sobersanta]

Thank you! Starting tests, will let you know if I find any new cases ;)

[akarnokd]

I'll merge this once the previous PR clears, then I will run it on the Build_Matrix to see if any flakyness arises.

[sobersanta]

I'm afraid I got another case of hang during normal sequence shutdown via onComplete.
Still trying to have a reproducing test case... From my logs I can see that eviction of group happened during shutdown sequence, that's all I got for now.

[akarnokd]

Thanks. No rush here with the release. If you find something, it will go in the next PR.

[sobersanta]

Oh, I'm so sorry. Seems to be a false alarm, I was a bit lost with versions!

[davidmoten]

Are the groupBy fixes something that should be backported to 2.x?

[akarnokd]

The abandonment can't be backported because that's a breaking behavior change. The group termination might be, but the unit tests in these 3.x fixes may hang anyway due to not having a properly overarching group state management. We'll only know if a backport is attempted. PR welcome.

[davidmoten]

Thanks @akarnokd. I haven't seem symptoms in our applications of problems so not urgent for me yet but I'll put it on our backlog.