Premature thread interruption when using Mono.just().transform(flux)

[dfeist]

See: ReactiveX/RxJava#5711
See: ReactiveX/RxJava#5715

In the following scenario:

Scheduler scheduler = Schedulers.fromExecutorService(Executors.newCachedThreadPool());
Flux<Integer> flux = Flux.just(1);
Function<Publisher<Integer>, Publisher<Integer>> addOne = f -> Flux.from(f)
		                                                    .map(i -> i + 1)
		                                                    .publishOn(scheduler);
flux.concatMap(j -> Mono.just(j).transform(addOne)).subscribe(System.out::println);

• the Mono.transform, given addOne is not a Mono uses MonoNext to wrap
• MonoNext does s.cancel() before actual.onNext(t);
• WorkerTask future gets cancelled (due to MonoNext s.cancel), and can get cancelled even before reactor.core.scheduler.WorkerTask#setFuture. If this happens thread is interrupted and processing on that thread (e.g. reading of IO) can be cut short

In my case the thread doing the map() is an HTTP selector thread and the Thread.interrupt() provoked in WorkerTask#setFuture causes the reading of AbstractInterruptibleChannel to be cut short with a ClosedByInterruptException resulting in an empty response body.

The cancel-before-next in MonoNext is necessary but causes a race condition between reactor.core.scheduler.WorkerTask#dispose and sr.setFuture(f) (in reactor.core.scheduler.Schedulers#workerSchedule), where when dispose() is called before setFuture(f) the thread that is still processing the onNext() signal is prematurely interrupted.

[simonbasle]

has-workaround: The particular code snippet can be changed to using Flux.just(f) in the concatMap instead of Mono.just(j), which will avoid the MonoNext and associated cancellation altogether.