Need help on backpressure and auto pause / resume during high load for reactive consumer

[pavanpkp33]

Hi All,

I have a reactive kafka consumer which consumes data from a topic and pushes it to an endpoint. However, these endpoints are flaky and are error prone. I have circuit breakers which trip on failures and I induce an artificial delay Mono.delay to slow down the pipeline. By theory, reactive backpressure should kick in and pause the kafka consumers right? I see that If i set my max.poll.interval.ms to low number, it does not pause the consumer and it results in rebalancing. How can I make sure that the Kafka consumer applies back-pressure and pauses the consumption until the downstream messages are done processing?
Here's my consumer code:

protected Disposable pollRecordsWithAutoCommit(KafkaReceiver<byte[], T> receiver) {
    return receiver
        .receiveAutoAck()
        .onErrorStop()
        .flatMap(
            consumerRecordFlux -> processFluxStream(consumerRecordFlux),
            jobConfiguration.getConsumerParallelPower())
        .onErrorContinue(
            // ALL uncaught exceptions will be published here.
            // TODO:: post to DLQ with retry count or possibly other meta data
            // `o` is the KAFKA record
            (throwable, o) -> {
              log.error(
                  "Error while processing record. [error:{}, record:{}]", throwable.toString(), o);
            })
        .subscribe();

processFluxStream has Mono.delay code to slow down incase of circuit breaker errors.

How do I go about linking the reactive pipeline to pause and resume on these signals and also avoid rebalancing issues?
I read about an earlier issue which had the same problem but that does not have a solution.

Thanks.

[simonbasle]

Which version of reactor-kafka are you using? Version 1.3.x was overhauled and might behave better in terms of backpressure pausing consumption. @garyrussell any insight ?

[garyrussell]

@simonbasle I have done some analysis (with 1.3.3-SNAPSHOT). Here is what I found...

        Disposable flux = receiver.receive()
            .doOnNext(rec -> {
                System.out.println(rec);
            })
            .subscribe();

If I set a breakpoint on the System.out, I see the main dispatcher thread hung (so polling is stopped). So...

        Disposable flux = receiver.receive()
            .publishOn(Schedulers.newSingle("mightHang"))
            .doOnNext(rec -> {
                System.out.println(rec);
            })
            .subscribe();

Now, the polling thread continues to poll. When the requests count down to 0 (due to the back pressure), we correctly pause the consumer.

However, due to this code....

reactor-kafka/src/main/java/reactor/kafka/receiver/internals/ConsumerEventLoop.java

Lines 243 to 247 in 1c67718

	if (isActive.get()) {
	if (r > 1 || commitEvent.inProgress.get() > 0) {
	schedule();
	}
	}

...we don't schedule the PollTask because r == 0 and so we stop polling.

It looks to me like we should always reschedule, even when r is 0.

When I change it to unconditionally schedule, it works as expected (keep polling while paused, returning 0 records on each poll).

When I release the breakpoint, we resume the consumer as expected (I was actually testing with my PR in place).

However, there is one more oddity.

Initially, requested is set to 256 and counts down to 64 before being replenished to 256.

After forcing the back pressure, it counts down to 0 and we pause the consumer.

When I released the break point and we resumed the consumer, requests went to 767, counted down to 744; then it went to 936, counting down to 744, with the latter sequence repeated over and over.

[pavanpkp33]

@simonbasle I'm using 1.3.0. @garyrussell any future releases planned to fix this? Thanks.

[garyrussell]

I will try to get a fix into 1.3.3, due in a couple of weeks.