ParallelFlowable, sequential and exception management

[marcosalis]

Hi,
I'm stuck on a problem which might or might not be an issue on the current version of the ParallelFlowable, or if it is a desired behaviour, it's not specified anywhere in the documentation.
In a few words, I'm having issues propagating errors from a ParallelFlowable to a single Flowable (when calling sequential()).
What I would expect is that any error generated during the execution of operators on the ParallelFlowable would be propagated to the subscriber on the Flowable.
What really happens is that the exception is not propagated, and it's thrown even if it's a checked exception.
Here is a code example:

Flowable.just(request1, request2)
                .parallel() // parallel execution is needed
                .runOn(Schedulers.io())
                .map(request -> request.execute()) // execute() can throw an exception
                .sequential()
                .toList()
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(this::success, this::failure);

failure(exception) is not called, and the exception is actually thrown.

I've also tried this and the exception is still not propagated to the sequential Flowable:

Flowable.just(request1, request2)
                .parallel() // parallel execution is needed
                .runOn(Schedulers.io())
                .flatMap(request -> {
                    try {
                        return Flowable.just(request.execute());
                    } catch (Exception e) {
                        return Flowable.error(e);
                    }
                }
                .sequential()
...

I'm using version 2.0.5 of RxJava on Android.

Thank you for your help.

[akarnokd]

The underlying problem is that sequential can only allow one onError. If multiple rails fail, one gets through and the others get sent to RxJavaPlugins.onError.

On one hand, this is a general property of RxJava 2 and requires you to setup a global error handler based on what you consider fatal and what not.

On the other hand, the internals of the parallel operators don't leave much option. Currently, if a rail fails it cancels the upstream rail (which cancels the original Flowable in the chain) and emits an onError. When the error reaches the sequential operator, the operator cancels the rest of the rails.

For a workaround, I suggest you switch back to a traditional parallel execution:

Flowable<T> a = Flowable.just(request1).subscribeOn(Schedulers.computation()).map(...);
Flowable<T> b = Flowable.just(request2).subscribeOn(Schedulers.computation()).map(...);

Flowable<T> c = Flowable.mergeArrayDelayError(a, b);

I need more time to figure out a better way to support your failure mode.

[marcosalis]

Thank you @akarnokd for the super fast turnaround on this!
I'll go with Flowable.mergeArrayDelayError() for now, since it seems to behave as expected, and I look forward to check out the new solution with ParallelFlowable.
In general, when it comes to parallel computations there are three common scenarios to handle a failure and it would be good if developers were able to choose among all of them:

• Make all the computations fail when one of them fails
• Notify all successful computations until the failure, then fail all the others
• Notify all successful computations, before and after the failed computation

[akarnokd]

See #5202.

I'm closing this issue due to inactivity. If you have further input on the issue, don't hesitate to reopen this issue or post a new one.