KafkaMessageListenerContainer pauses for a long time after error handling

[antonin-arquey]

In what version(s) of Spring for Apache Kafka are you seeing this issue?

>= 2.9.0

Describe the bug

It is stated in the documentation that

Starting with version 2.9, the DefaultErrorHandler can be configured to provide the same semantics as seeking the unprocessed record offsets as discussed below, but without actually seeking. Instead, the records are retained by the listener container and resubmitted to the listener after the error handler exits (and after performing a single paused poll(), to keep the consumer alive;

Listeners that receives a large amount of data will start to lag behind quickly if they encounter unexpected errors during processing because they are constantly paused and polling for a long time (by default the poll timeout is 5 seconds but could be larger if set otherwise by the user).

To Reproduce

Really easy to test, just throw any exception in a listener configured with a CommonLoggingErrorHandler and a poll timeout set to any large value.

The listener will be stuck for the duration of the poll timeout before processing the remaining records.

Expected behavior

Do not poll with the consumer paused before processing the remaining records (or at least poll with a shorter poll timeout) as it is not expecting to receive a record anyway.

This should probably be mentionned in the release notes as it seems small but is quite a big behavoural change and can lead to issues for some people upgrading to SpringBoot 3.

[garyrussell]

a big behavoural change

??

The option is turned off by default for exactly that reason.

Do not poll with the consumer paused ...

We have to poll, to avoid a possible rebalance when there are many errors in the records returned by the previous poll or the aggregate back offs exceed max.poll.interval.ms.

But your point is well taken. We should certainly reduce the poll time to zero (or near it) when we know that all partitions are paused.

In a future release, we should consider reducing the default poll timeout.

[antonin-arquey]

??

The option is turned off by default for exactly that reason.

Sorry you are right, the DefaultErrorHandler does not behave this way so in fact there is no behavioural change out of the box.

The issue we faced is that we used the CommonLoggingErrorHandler that replaced the previously existing LoggingErrorHandler in version 2.8.

Then when we tried the upgrade to springboot 3, we encountered the issue as we pulled in the seeking change stated by the documentation.

It was weird for us as it is the most 'harmless' of error handlers and we did not expect any changes from the previous one.

Maybe the misunderstanding comes from the default implementations in the CommonErrorHandler interface that enable this behaviour by default for error handlers that do not override them:

/**
* Return false if this error handler should only receive the current failed record;
* remaining records will be passed to the listener after the error handler returns.
* When true (default), all remaining records including the failed record are passed
* to the error handler.
* @return false to receive only the failed record.
* @deprecated in favor of {@link #seeksAfterHandling()}.
* @see #handleRecord(Exception, ConsumerRecord, Consumer, MessageListenerContainer)
* @see #handleRemaining(Exception, List, Consumer, MessageListenerContainer)
*/
@Deprecated(since = "2.9", forRemoval = true) // in 3.1
default boolean remainingRecords() {
return false;
}

/**
* Return true if this error handler performs seeks on the failed record and remaining
* records (or just the remaining records after a failed record is recovered).
* @return true if the next poll should fetch records.
*/
@SuppressWarnings("deprecation")
default boolean seeksAfterHandling() {
return remainingRecords();
}

[garyrussell]

Oh; sorry; yes, the default should have it turned off too.

[garyrussell]

However, the logging handler handles (logs) the error, so there should be no pause at all; so that is a bug too if that is indeed what's happening.

It shouldn't happen though because the handled boolean is true if the handler doesn't throw an exception.

[garyrussell]

The bug is here:

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

Lines 2921 to 2933 in 0cad2d7

	if (!handled) {
	records.computeIfAbsent(new TopicPartition(cRecord.topic(), cRecord.partition()),
	tp -> new ArrayList<ConsumerRecord<K, V>>()).add(cRecord);
	}
	while (iterator.hasNext()) {
	ConsumerRecord<K, V> next = iterator.next();
	records.computeIfAbsent(new TopicPartition(next.topic(), next.partition()),
	tp -> new ArrayList<ConsumerRecord<K, V>>()).add(next);
	}
	if (!records.isEmpty()) {
	this.remainingRecords = new ConsumerRecords<>(records);
	this.pauseForPending = true;
	}

The while loop should be with the if (!handled) test.

Thanks for reporting.