Flux.replay / Flux.cache hangs or serves wrong values due to regressions in 3.3.[0-2]

[simonbasle]

In 3.3.0, #1185 introduced a change that aims at performing the minimum necessary request on a cached/replayed source upon connect. Comparing to Californium where the source would always be requested by Long.MAX_VALUE upon connection, 3.3.0 would attempt to take the maximum requested amount from early subscribers, ie these that subscribed to the ConnectableFlux before connect() is called.

There was several bugs with that new implementation, that led to underrequesting and hanging:

• Pre-connect subscribers would participate in the initial request, but if they request in "batches" their subsequent requests wouldn't get propagated (issue Flux.cache hangs indefinately in multi-batch flow scenarios (many request calls). #1921, fixed in fix #1921 Propagate all req from FluxReplay pre-connect subscribers #1948)
• With that first change, having no early subscriber before connect() means no request was made at all, instead of falling back to the MAX_VALUE behavior (causes ConnectableFlux changes behavior between 3.2 and 3.3 #2028 and Cached Flux problem with application/stream+json response type spring-projects/spring-framework#24472, fixed in fix #2030 Let replay request(UNBOUNDED) when no early subscriber #2031)