Flowable.timeout() signals TimeoutException during backpressure

[davidmcote]

The Flowable.timeout() operator signals TimeoutException when:

• EITHER the upstream is emitting too slowly (expected)
• OR if the downstream is requesting the upstream to emit items too slowly. (surprising)

Here's a toy example demonstrating TimeoutException during backpressure:

    @Test
    public void testTimeout() {
        Flowable
                .just("Hello, world!")
                .repeat(100)
                .timeout(1, TimeUnit.SECONDS)

                .blockingSubscribe(new Subscriber<>() {
                    Subscription sub;

                    private void slowRequest(final long n) {
                        Schedulers.io().scheduleDirect(() -> sub.request(n), 5, TimeUnit.SECONDS);
                    }

                    @Override
                    public void onSubscribe(final Subscription sub) {
                        this.sub = sub;
                        slowRequest(1);
                    }


                    @Override
                    public void onNext(final String o) {
                        System.out.println(o);
                        slowRequest(1);
                    }

                    @Override
                    public void onError(Throwable throwable) {
                        System.out.println(throwable);
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("COMPLETE");
                    }
                });
    }

Result:

java.util.concurrent.TimeoutException: The source did not signal an event for 1 seconds and has been terminated.

The error message indicates that the source hadn't signal for longer than the timeout, which is technically accurate, but the source was entirely capable of emitting as much as the downstream subscriber wanted.

Should the family of timeout() operators be taking into account whether the downstream has actually requested more items? Ie: don't timeout backpressure?

(Apologies if this has already been asked or if I have overlooked another operator with the desired behavior.)

[akarnokd]

The operator doesn't consider why there was no next item within the allotted time. If there was none, errors. Works as intened.

Pretty much everything should be issuing requests as soon as possible. The worst case could be that you have a so low threshold and so long chain with async requesting that you can get timed out just by subscribing.

In any case, getting a timeout means you have to investigate and remove the source of the disruption.

Why do you have the request delay issue?

[davidmcote]

The request delay was an artificial example to demonstrate backpressure. The aim was to model a slow downstream consumer.

As another toy example that is perhaps more realistic in practice:

    @Test
    public void testTimeout() {
        Flowable
                .just("Hello, world!")
                .repeat(100)
                .timeout(1, TimeUnit.SECONDS)

                .concatMapSingle(request -> fakeSlowRpcCall(request))

                .blockingSubscribe(); // Throws TimeoutException
    }

    private static Single<String> fakeSlowRpcCall(final String request) {
        return Single
                .just("Response to " + request)
                .delay(5, TimeUnit.SECONDS);
    }

In this scenario, I would not have expected the timeout() operator to care how long the downstream fakeSlowRpcCall() took to complete.

The motivation behind this ask is that I wish to impose a minimal producer rate on my datasource if it's too slow, but to continue honoring backpressure without placing any lower bound on the downstream processing speed. Eg:

private static Flowable<Datum> dataProducer() {
   return Flowable
                   ...
                .timeout(1, TimeUnit.SECONDS);
}

private static Disposable dataConsumer(final Flowable<Datum> upstream) {
    return upstream
        .concatMapSingle(...slow sequential per-item work here..)
        .subscribe();
}