Reactor-core: IllegalStateException: Iterating over a toIterable() / toStream() is blocking

[anuchandy]

The following code:

Flux.fromIterable(Flux.fromArray(new Integer[] {1, 2, 3}).toIterable())
        .subscribeOn(Schedulers.single())
        .doOnNext(i -> System.out.println(i))
        .blockLast();

throws the below error:

>
Exception in thread "main" java.lang.IllegalStateException: Iterating over a toIterable() / toStream() is blocking, which is not supported in thread single-1

The same error is reproducible with .subscribeOn(Schedulers.parallel()), but not with elastic Scheduler. What could be causing this?

Reactor-core version: 3.3.0-RELEASE
Machine: MacOS Mojave Version 10.14.6

[UgiR]

"Transform this Flux into a lazy Iterable blocking on Iterator.next() calls."
You are using a blocking operation in a thread marked as non-blocking.

Why create a Flux from an array, turn it into an iterable, and then create a new Flux from that iterable? Wouldn't it make sense to just create your Flux directly from the array? Or if you need to create a flux from another publisher, you can use Flux.from() -- no need for the intermediate iterable.

[conniey]

A model class in our repository returns an iterable constructed from a Flux. However, how we expose it to users in a Flux, so it ends up breaking.

I'm trying to understand how this behaviour can be fixed. And why it's doing what it's doing.

https://github.com/Azure/azure-sdk-for-java/issues/6515

[UgiR]

See section 4.5 of the documentation:

While boundedElastic is made to help with legacy blocking code if it cannot be avoided, single and parallel are not. As a consequence, the use of Reactor blocking APIs (block(), blockFirst(), blockLast() (as well as iterating over toIterable() or toStream()) inside the default single and parallel schedulers) results in an IllegalStateException being thrown.

When you convert your Flux to a stream or iterable, you are leaving the Flux/publisher universe and going back to synchronous code.

If you try to go back to a Flux from the iterable, hasNext() blocks until an element is available, and you cannot have blocking operations on those Schedulers' threads.

You want the flux to be 'end-to-end'. Since you are exposing to users as a Flux, you probably don't want to reduce it to a stream or iterable before it gets there.

[simonbasle]

as @UgiR said, this restriction is a general one that prevents synchronous code to execute on shared schedulers, potentially impacting other reactive parts of your application.

To be 100% transparent, fromIterable is in a unique situation among the from operators where it can still revert the effects of toIterable (because the underlying Publisher is still reachable and isn't subscribed to until iteration starts). But I don't think covering this special case is worth it because it would open the door to more puzzlement as to why it works for from/toIterable and not from/ToFuture or from/ToStream.

So for now, please try your hardest to avoid exposing the intermediate Iterable and converting it back to Flux using fromIterable.