@Test
public void create() throws ExecutionException, InterruptedException {

    Flowable<Integer> flowable = Flowable.fromArray(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
            .subscribeOn(Schedulers.computation());
    AtomicReference<Throwable> error = new AtomicReference<>();
    ReactiveStreams.fromPublisher(flowable)
            .filter(i -> i < 4)
            .map(this::squareOrFailed)
            .onError(error::set)
            .map(this::asString)
            .toList()
            .run().toCompletableFuture().exceptionally(x -> Collections.emptyList()).get();
    assertThat(error.get()).hasMessage("failed");
}