Exception in the thread who subscribed the flux or mono

[stavalfi]

The following scenario will prove theoretically that even if I check if the stream canceled before sending error signal, then flux will throw an exception to the subscribed thread: consider the following scenario inside Flux.create which the main thread subscribed to it (flux.create(..)....parallel()...take(1)...):

1. main thread check if the stream is canceled before sending an error signal. the answer is no.
2. main thread is on hold by the operating system.
3. parallelThread1 pass the take(1) and terminate the stream.
4. parallelThread1 is on hold by the operating system.
5. main thread is alive now and sending an error signal. the error will be thrown in the main thread.

In general, what are my options to avoid exceptions in the thread who subscribed the flux?

[dfeist]

@smaldini I believe this also fixes the minor inconsistency we talked about the other day.

[simonbasle]

@stavalfi the PR that @smaldini opened (#1085) replaces the onErrorDropped behaviour with an onOperatorError behaviour. The difference is that, unless the global hook has been customized, the error will now be swallowed instead of being thrown. For create, one could argue this is a saner default.

Additionally, even though it was also the case with the previous behaviour, you can customize the hook at the level of a given Flux/Mono rather than globally and then catch the dropped errors. This necessitates a bit of undocumented boilerplate:

Flux.create(...)
    .subscriberContext(Context.of("reactor.onOperatorError.local", 
        (droppedError, valueIfRelevant) -> {...process dropped error...});

Would you be able to quickly use a 3.2.0.BUILD-SNAPSHOT once that PR gets merged to see if that improves your situation, and provide feedback? We would then maybe consider porting it back to the 3.1.x line (as we'll probably release a 3.1.5).

[simonbasle]

☝️merged into master and should be available in a snapshot soon, pending user feedback before we decide on a potential backport to 3.1

[stavalfi]

Hi team. thanks for the quick upgrade.

1. I will try it out until the end of this week. If it is critical for you to make any decision, please feel free to ignore my feedback.

2. I was asking if it is thread-safe to use FluxSink::isCancelled because as I mentioned above, there can be a situation when FluxSink::isCancelled returns false but only then the stream is canceled.

For solving that, I can see the following fix:
FluxSink::isCancelled will receive a lambda which will be activated after you will lock an object which will block any attempt to cancel the stream for any reason. The disadvantage is clear: if the lambda I provided will not end, the stream will never be canceled.

Thanks again!

[stavalfi]

My feedback: working great. When will this feature be added to 3.1?

[simonbasle]

Great @stavalfi. This has now been backported in the 3.1.x branch. The 3.1.5 release should happen within the month.