Fragile window abandonment during concatMap operations; silent data loss

Inizia eseguendo le due riproduzioni usando Flowable.window() con concatMapSingle() e concatMapMaybe(), confrontando le varianti di blockingSubscribe(). Traccia l’interazione tra l’abbandono delle finestre e la sottoscrizione downstream; il lavoro è completato quando le finestre non vengono eliminate silenziosamente oppure l’abbandono produce l’errore documentato.

Combining the Flowable.window() family of operators with any of the Flowable.concatMap*() operators appears to cause window abandonment bugs which are sensitive to how the outer Flowable is processed downstream. This despite the given concatMap lambdas returning an Rx Single/Maybe ready to be subscribed as soon as they are returned and which will signal the upstream window for subscription synchronously.

In the demonstrations below, I subscribe to identical .window().concatMap() constructions with very slightly different downstreams (.blockingSuscribe(System.out::println) vs. .doOnNext(System.out::println).blockingSubscribe()) and they emit drastically different results due to silent data-loss.

Example: Windowing a stream into lists, similar to the buffer() operator.

    public static void main(String[] args) {
        final var content = Flowable
                .rangeLong(0, 100)
                .window(10)
                .concatMapSingle(Flowable::toList);

        System.out.println("Example 1: Expected behavior for concetMapSingle(Flowable::toList)");
        content
                .doOnNext(System.out::println)
                .blockingSubscribe();

        System.out.println("Example 2: Window abandonment & data-loss");
        content
                .blockingSubscribe(System.out::println);
    }

Output:

Example 1: Expected behavior for concetMapSingle(Flowable::toList)
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
[10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
[20, 21, 22, 23, 24, 25, 26, 27, 28, 29]
[30, 31, 32, 33, 34, 35, 36, 37, 38, 39]
[40, 41, 42, 43, 44, 45, 46, 47, 48, 49]
[50, 51, 52, 53, 54, 55, 56, 57, 58, 59]
[60, 61, 62, 63, 64, 65, 66, 67, 68, 69]
[70, 71, 72, 73, 74, 75, 76, 77, 78, 79]
[80, 81, 82, 83, 84, 85, 86, 87, 88, 89]
[90, 91, 92, 93, 94, 95, 96, 97, 98, 99]

Example 2: Window abandonment & data-loss
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
[10]
[20]
[30]
[40]
[50]
[60]
[70]
[80]
[90]

Example: Windowing a stream into a simple reduction operation.

    public static void main(String[] args) {
        final var windowSums = Flowable
                .rangeLong(0, 100)
                .window(10)
                .concatMapMaybe(window -> window.reduce(Long::sum));

        System.out.println("Example 3: Expected behavior for concetMapMaybe(window -> window.reduce(Long::sum))");
        windowSums
                .doOnNext(System.out::println)
                .blockingSubscribe();

        System.out.println("Example 4: Silent data loss");
        windowSums
                .blockingSubscribe(System.out::println);
    }

Output:

Example 3: Expected behavior for concetMapMaybe(window -> window.reduce(Long::sum))
45
145
245
345
445
545
645
745
845
945

Example 4: Silent data loss
45
10
20
30
40
50
60
70
80
90

The flatMap*() family of operators seem less prone to this behavior but don't guarantee to retain the upstream ordering.

Ideally, I'd expect window() to be compatible with concatMap(). But if such is impossible, I'd expect the window-abandonment detection described in https://github.com/ReactiveX/RxJava/wiki/What%27s-different-in-3.0#window-abandonment-in-window to at least signal an error.