search

reactor / reactor-core

Non-Blocking Reactive Foundation for the JVM
http://projectreactor.io
Apache License 2.0
4.87k stars 1.18k forks source link

Flux.mergeSequential does not subscribe to last Producer in specific circumstances #3802

Open jmmerz opened 2 months ago

jmmerz commented 2 months ago

Problem Description

I'm upgrading an application to take Reactor from 3.4.x to the latest (3.6.5) and have run into this problem:

If I invoke Flux.mergeSequential(Publisher<? extends I>... sources) like:

Flux<String> = Flux.mergeSequential(producer1, producer2, ..., producerN, monoFinal);

where producer1, producer2, ..., producerN have code that prepares an operation and then awaits a signal, and monoFinal is a Mono.fromCallable() containing code that sends the signal to allow the code run by release the first N Producers to continue and complete.

It seems the requirements to trigger the unexpected behavior are:

  1. The first N Producers must begin upon subscription and wait for a signal from the last one.
  2. The last Producer must be a Mono.fromCallable(). I've found that if I use a Mono.fromRunnable() instead, the contained Runnable is executed, while in the former case, the Callable is not executed.

Background (my specific use case for the above)

I have some code structured as described above where producer1, producer2, ..., producerN send commands to a Redis client that is configured not to flush commands over the wire immediately, and monoFinal is a Mono.fromCallable() that tells the client to flush the commands so that they are sent over the wire as a batch after the last Mono runs.

When invoked like this, producer1, producer2, ..., producerN are all subscribed by the merged Flux, and their code waits for the signal sent by monoFinal, but monoFinal is not subscribed - at least its Callable is never executed.

I've developed a simplified test case not requiring external services/libraries to demonstrate what's happening (link below in Steps to Reproduce).

Expected Behavior

The merged Flux should subscribe to all Producers eagerly and their emitted values should be produced in order.

Actual Behavior

The first N Producers are subscribed, but the last Mono appears never to be subscribed (or at least it's Callable is never never executed).

Steps to Reproduce

I've created a small project with a test case to Github. If it is preferable to deliver test code in another way, please let me know, I'm happy to oblige. See the project in test-reactor-flux-mergeSequential:

Possible Solution

I don't have a solution, but I've found a workaround (explained below) in case it helps.

Workaround 1

If I redefine monoFinal using Mono.fromRunnable() in place of the existing Mono.fromCallable(), things work as expected.

Workaround 2

(This workaround seems less good, but I'm leaving it documented in case it helps narrow down the cause.)

If I invoke Mono.doOnSubscribe(Consumer<? super Subscription> onSubscribe) on monoFinal even with a No-Op Consumer, it causes things to work.

The same thing happens with Mono.log() which I suspect is due to adding signal handling for onSubscribe as well. Other error handlers (onCancel, onError, onTerminate) have no effect.

Your Environment

jmmerz commented 2 months ago

Update: I've discovered that the problem appears related specifically to using a Mono.fromCallable() as the final producer for the Flux.mergeSequential(). If I instead use a Mono.fromRunnable() the code works as expected.

I've updated the description and workarounds in the original ticket accordingly.