Reactor-core: WindowTimeout backpressure

[bsideup]

Expected behavior

windowTimeout respects the backpressure

Actual behavior

Not verified if expected or not:
The receiver is overrun by more signals than expected (bounded queue...)

Steps to reproduce

Flux
        .generate(sink -> {
            System.out.println("generate");
            sink.next(UUID.randomUUID().toString());
        })
        .log("source", Level.INFO, SignalType.REQUEST)
        .limitRate(5)
        .windowTimeout(10, Duration.ofSeconds(2))
        .concatMap(batch -> batch.delayElements(Duration.ofMillis(200)))
        .blockLast()

Reactor Core version

3.1.4.RELEASE

JVM version (e.g. java -version)

Any

[alex-lx]

Meet the same issue when implementing a Kafka consumer which bulk save the records to another database.

[bsideup]

@alex-lx I had exactly the same source - Kafka :)

FYI here is a workaround for our use-case:
https://github.com/bsideup/liiklus/blob/2184d62d400a6ab72775395ba3f1a1eca9f64910/examples/java/src/main/java/com/example/Consumer.java#L67-L72

Basically, just do:

.window(size)
.concatMap(
    batch -> batch
        .delayUntil(process)
        .sample(windowDuration)
        .delayUntil(closeSubWindow),
    1
)

This way the window will be closed either after receiving size items or by delay thanks to .sample(), and it will respect the backpressure

[dforsl]

Hi, is there any plan on looking into this for windowTimeout and/or bufferTimeout?

There are a few questions around this on SO and in the gitter channel, but I haven't found any answers other than something along the lines of "not supported right now". It would be great to know if it's on the horizon, or if this simply isn't usage intended to be covered by reactor.

For instance, it would be great when reading from Kafka and persisting the records to a database in batches. As it is now, that doesn't seem possible out-of-the-box without the risk of an overflow-error if the process of writing to the database was temporarily slowed down for whatever reason.

The above work-around looks great for when you're processing records one by one, but I haven't been able to come up with anything for batching. Combining limitRate and onBackpressureBuffer isn't very attractive either, as it kind of cancels the backpressure-mechanism that is typically present when writing.

I've mainly looked into bufferTimeout, where I suppose the issue is that it is incapable of actually taking care of the result of the demand that itself has created. Would it be possible to make it buffer whatever that it has requested from upstream, when downstream isn't yet ready? If not always, then by specifying an overflow-strategy so that I can configure it to buffer? I fully accept that any additional items than those requested would cause an overflow-error, but I don't think it makes sense for records that the operator has requested itself just because they arrived after the time-boundary.

Essentially, what I'd want is being able to do, without risking an overflow-error on a temporary slow-down when writing to the database, is simply something like:

  .bufferTimeout(1000, Duration.ofMillis(500))
  .concatMap { batch ->
    val result: Mono<Void> = // insert the batch
    return result.then(Mono.defer { //acknowledge offsets })
  }

Any input on this, @simonbasle @bsideup ? If you have ideas on how this can be solved by other operators, then I'd love to hear about those!

I hope it wasn't that bad of me to put it in here, I can open a new ticket if preferred! I just happened to find this old one. I really enjoy reactor, why I'm really keen on seeing if this is doable.
/d

[bsideup]

The workaround is for the use case and not for the reported issue. The issue remains and hurts really hard.

[Ksnz]

Any ideas for workarrounds?

[asyedin]

any chance this is going to be prioritized?

[mprowaznik]

This issue is very frustrating. And not fixed since 2018 🤦

[hepin1989]

sbt:MyLearning> runMain back.ReactorBuffer
[info] Compiling 1 Java source to /Users/kerr/IdeaProjects/untitled1/MyLearning/target/scala-2.13/classes ...
[info] Done compiling.
[warn] Multiple main classes detected.  Run 'show discoveredMainClasses' to see the list
[info] running back.ReactorBuffer 
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
generate
[error] (run-main-1) reactor.core.Exceptions$OverflowException: The receiver is overrun by more signals than expected (bounded queue...)
[error] reactor.core.Exceptions$OverflowException: The receiver is overrun by more signals than expected (bounded queue...)
[error]     at reactor.core.Exceptions.failWithOverflow(Exceptions.java:221)
[error]     at reactor.core.publisher.FluxWindowTimeout$WindowTimeoutSubscriber.onNext(FluxWindowTimeout.java:233)
[error]     at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:203)
[error]     at reactor.core.publisher.FluxGenerate$GenerateSubscription.next(FluxGenerate.java:169)
[error]     at back.ReactorBuffer.lambda$main$0(ReactorBuffer.java:18)
[error]     at reactor.core.publisher.FluxGenerate.lambda$new$1(FluxGenerate.java:56)
[error]     at reactor.core.publisher.FluxGenerate$GenerateSubscription.fastPath(FluxGenerate.java:223)
[error]     at reactor.core.publisher.FluxGenerate$GenerateSubscription.request(FluxGenerate.java:202)
[error]     at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.request(FluxPeekFuseable.java:137)
[error]     at reactor.core.publisher.FluxWindowTimeout$WindowTimeoutSubscriber.onSubscribe(FluxWindowTimeout.java:177)
[error]     at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onSubscribe(FluxPeekFuseable.java:171)
[error]     at reactor.core.publisher.FluxGenerate.subscribe(FluxGenerate.java:83)
[error]     at reactor.core.publisher.Flux.subscribe(Flux.java:8325)
[error]     at reactor.core.publisher.Flux.blockLast(Flux.java:2496)
[error]     at back.ReactorBuffer.main(ReactorBuffer.java:24)
[error]     at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
[error]     at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
[error]     at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
[error]     at java.base/java.lang.reflect.Method.invoke(Method.java:567)
[error] stack trace is suppressed; run last Compile / bgRunMain for the full output
[error] Nonzero exit code: 1
[error] (Compile / runMain) Nonzero exit code: 1
[error] Total time: 1 s, completed 2020年5月7日 上午12:41:45
[IJ]sbt:MyLearning> 

[vsethi13]

Got hit by this one today! Frustrating :/

[bsideup]

To everyone affected by this issue:
Please share your use case! We want to introduce an improvement that should solve the problem, but need to understand the use cases better, so that it covers as much as possible.

Thanks in advance!

/cc @mprowaznik @vsethi13 @dforsl @alex-lx

[sneakythr0ws]

To everyone affected by this issue:
Please share your use case! We want to introduce an improvement that should solve the problem, but need to understand the use cases better, so that it covers as much as possible.

Thanks in advance!

/cc @mprowaznik @vsethi13 @dforsl @alex-lx

.bufferTimeout(..).concatMap() +1

[bsideup]

@sneakythr0ws well, this is code. But maybe your use case can be done differently :) Hence the question about the use case.

[sneakythr0ws]

@sneakythr0ws well, this is code. But maybe your use case can be done differently :) Hence the question about the use case.

I use bufferTimeout and concatMap after it for batch insert. In my case I can replace it with buffer()

[dforsl]

@bsideup I'll try to descibe a, for us, common processing flow:

1. Read events from Kafka
2. Save a batch of events to a database, that'll eventually form a "complete" record
3. Commit offsets
4. "Complete" records are sent downstream for further processing

We have been using reactor-kafka (altough it has proven somewhat unstable with respect to error handling). Generally,
we split the flow into separate rails at _1_ by grouping on partition. For each rail, we batch the incoming events, up to some limit, before do a batch-insert to our database. This enables us to handle both real-time peek hours as well reprocess a topic, when needed, in a timely manner.

For latency reasons (as our load various throughout the day) we also want to limit the batching with respect to time. But the different times we've researched this, we haven't been able to find a solution with reactor where we can do this and respect backpressure. As we can make fairly good guesses if an event will make a record complete, we could maybe utilise that "somehow". But that seems to grow complicated pretty quickly as opposed to just "batch with timeout".

I'm sure there are better ways to structure the processing and so on. But because we need to use a database as intermediate step to store partial records, as we'd otherwise have to start from way too far back on startup due to how long it can take before a record is complete, we seem to be somewhat stuck in trying to solve this with reactor.

In an attempt to generalise, the key characteristics to me are:

• batch-op towards an external service
• latency requirements
• respect backpressure

[chibenwa]

As requested I share my use case: apply a throttling to limit background operations resource consumptions. As part of the Apache JAMES mail server, some "background tasks" can be specified a "messagesPerSecond" parameters.

Our initial "throttling" implementation was:

Function<Flux<T>, Flux<U>> throttling = flux -> flux.windowTimeout(elements, duration)
    .zipWith(Flux.interval(DELAY, duration))
    .flatMap(Tuple2::getT1, elements, elements)
    .flatMap(operation, elements)

As demonstrated in https://github.com/linagora/james-project/pull/3525 , it looks like we did not need windowTimeout and that window was enough (and did respect back-pressure) - I must admit I don't fully understand why slow sources are correctly handled though....

[bsideup]

FYI we just merged #2202 that allows doing "one-by-one" window processing (will be included in Reactor 3.4.0.M2, but you can try the SNAPSHOTs today)

It does not solve WindowTimeout, but at least we can now window + ACK every N seconds without loading unnecessary records into the memory, see this example:
https://twitter.com/bsideup/status/1278268312808493062?s=20

[dforsl]

@bsideup that sounds absolutely fantastic!

[lbilger]

My use case: Receive messages from Kafka, collect them in batches (using bufferTimeout in my case) and process each batch, applying a retry if an error occurs. So processing a batch takes until it completes successfully and eventually an overflow exception occurs if a lot of messages come in and some errors occur.

I _think_ I have found a workaround, but am not sure if I'm not overlooking something. So I'm sharing this to get some feedback as much as hoping to help others. It's written in Kotlin, but I think Java developers should be able to read it. The main idea is to use a Semaphore to limit the count of "in-flight" items (items forwarded to the bufferTimeout operator but not yet processed by the rest of the stream):

fun <T> Flux<T>.bufferTimeoutWithBackpressure(maxSize: Int, maxTime: Duration, maxInFlightElements: Int = (maxSize * 4).coerceAtLeast(Queues.SMALL_BUFFER_SIZE), remainingSteps: Flux<List<T>>.() -> Flux<List<T>>): Flux<List<T>> =
    this
        .concatMap {
            Mono.subscriberContext()
                .map { context -> context.get("semaphore") as Semaphore }
                .doOnNext { semaphore -> if (!semaphore.tryAcquire()) throw RuntimeException() }
                .onErrorStop()
                .retryBackoff(Long.MAX_VALUE, Duration.ofMillis(1), Duration.ofMillis(100))
                .thenReturn(it)
        }
        .bufferTimeout(maxSize, maxTime)
        .onBackpressureBuffer(maxInFlightElements)
        .run(remainingSteps)
        .concatMap { list ->
            Mono.subscriberContext()
                .map { it.get("semaphore") as Semaphore }
                .doOnNext { semaphore -> semaphore.release(list.size) }
                .thenReturn(list)
        }
        .subscriberContext(Context.of("semaphore", Semaphore(maxInFlightElements)))

This is the test I used to try it:

class BufferTimeoutWithBackpressureTest {
    companion object : KLogging()

    @Test
    fun `bufferTimeoutWithBackpressure works as intended`() {
        Flux.fromIterable(1..1000)
            .delayElements(Duration.ofMillis(1))
            .bufferTimeoutWithBackpressure(5, Duration.ofMillis(2)) {
                concatMap {
                    Mono.delay(Duration.ofMillis(20))
                        .thenReturn(it)
                }
            }
            .concatMapIterable { it }
            .reduce(0) { lastSeen, new ->
                assertThat(new).isEqualTo(lastSeen + 1)
                if (new % 100 == 0) {
                    logger.info { "Got $new" }
                }
                new
            }
            .block()
    }

    @Test(expected = IllegalStateException::class)
    fun `bufferTimeout does not work`() {
        Flux.fromIterable(1..1000)
            .delayElements(Duration.ofMillis(1))
            .bufferTimeout(5, Duration.ofMillis(2))
            .concatMap {
                Mono.delay(Duration.ofMillis(20))
                    .thenReturn(it)
            }
            .blockLast()
    }
}

[yakovs83]

I had a case very similar to @dforsl where flow from Kafka varies through the day, so I can't just use buffer() because some "leftover" elements may get stuck there waiting for more stuff to fill the buffer. bufferTimeout() does not respect backpressure, which is understandable, and blows up with a slow consumer, coupled with .onBackpressureBuffer() it is doing better but we take a hit in terms of memory in case of slow consumer.

The thing I came up with is something along the lines of

myFlux.timed().windowUntilChanged(periodic function based on elapsed time).concatMap(w -> w.map(Timed::get).bufferTimeout(size, timeout))

The downside is that I have to attach timing data and then remove it. Seems to work, at least better than bufferTimeout().onBackpressureBuffer() combo. Before that I came up with a convoluted recursive solution that would cache() the window then would try to collectList() it, timeout and repeat recursively, splitting original window in two. Recursion would terminate when our window size was 1. That looked good on paper, but didn't play nice memory-wise for some reason.

What I was looking for is a bufferTimeout-like operator, that would use timeout part to bracket the elements - something that would collect N elements or T seconds, but would not forcefully emit on a timer, but rather wait for a request - the time window would start after the accumulated buffer emitted (plus prefetch and such)

Edit: after some more thinking I came up with simpler windowing idea - wrap stream values in an Option (I like vavr one instead of standard Optional), merge with stream of Option.none() values that is generated via mapping over Flux.interval() and then I do .windowUntil(Option::isEmpty) windowing, filter empty Options, unwrap from Option, and slice resulted windows into buffers of the size I need.