With a @KafkaListener that returns a Mono, the error handler does not retry

[moonyoungCHAE]

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

3.3.7

Describe the bug

When creating an @KafkaListener that returns a Mono, the error handler does not retry when an exception occurs. I got this bug from stack overflow report.

To Reproduce

@Component
public class MyListener {
    @KafkaListener(
            clientIdPrefix = "myNewClient",
            groupId = "aa",
            topics = "monoTest"
    )
    public void listen(ConsumerRecord<?, ?> record) throws MyException {
        System.out.println("get data:  offset" + record.offset());
        throw new MyException();
    }
}

@Configuration
public class Config {
    private static final Logger logger = LoggerFactory.getLogger(Config.class);

    @Bean
    public CommonErrorHandler handler() {
        DefaultErrorHandler errorHandler = new DefaultErrorHandler(new FixedBackOff(1L, 20L));
        errorHandler.addRetryableExceptions(MyException.class, TimeoutException.class, ListenerExecutionFailedException.class);
        errorHandler.setRetryListeners((record, ex, attempt) ->
                logger.info("[CUSTOM-RETRY] attempt={} topic={} partition={} offset={} ex={}",
                        attempt, record.topic(), record.partition(), record.offset(), ex.toString())
        );
        return errorHandler;
    }
}

Expected behavior

When an exception occurs, the message is retried. For non-Mono listeners, retries behave as shown below.

2025-12-04T16:07:54.241+09:00  INFO 9045 --- [| adminclient-2] o.apache.kafka.common.metrics.Metrics    : Metrics reporters closed
get data:  offset0
2025-12-04T16:07:54.246+09:00  INFO 9045 --- [ntainer#0-0-C-1] org.example.consumer.Config              : [CUSTOM-RETRY] attempt=1 topic=monoTest partition=0 offset=0 ex=org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'public void org.example.consumer.MyListener.listen(org.apache.kafka.clients.consumer.ConsumerRecord<?, ?>) throws org.example.consumer.MyException' threw exception
2025-12-04T16:07:54.257+09:00  INFO 9045 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.LegacyKafkaConsumer  : [Consumer clientId=myNewClient-0, groupId=aa] Seeking to offset 0 for partition monoTest-0
2025-12-04T16:07:54.258+09:00  INFO 9045 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : Record in retry and not yet recovered
get data:  offset0
2025-12-04T16:07:54.742+09:00  INFO 9045 --- [ntainer#0-0-C-1] org.example.consumer.Config              : [CUSTOM-RETRY] attempt=2 topic=monoTest partition=0 offset=0 ex=org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'public void org.example.consumer.MyListener.listen(org.apache.kafka.clients.consumer.ConsumerRecord<?, ?>) throws org.example.consumer.MyException' threw exception
2025-12-04T16:07:54.754+09:00  INFO 9045 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.LegacyKafkaConsumer  : [Consumer clientId=myNewClient-0, groupId=aa] Seeking to offset 0 for partition monoTest-0
2025-12-04T16:07:54.755+09:00  INFO 9045 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : Record in retry and not yet recovered
get data:  offset0

But when the listener returns a Mono, the message is not retried as shown below.

get data:  offset0
2025-12-04T16:12:40.109+09:00  INFO 10101 --- [ntainer#0-0-C-1] org.example.consumer.Config              : [CUSTOM-RETRY] attempt=1 topic=monoTest partition=0 offset=0 ex=org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'public reactor.core.publisher.Mono<java.lang.Void> org.example.consumer.MyListener.listen(org.apache.kafka.clients.consumer.ConsumerRecord<?, ?>) throws org.example.consumer.MyException' threw exception
2025-12-04T16:12:40.122+09:00  INFO 10101 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.LegacyKafkaConsumer  : [Consumer clientId=myNewClient-0, groupId=aa] Seeking to offset 0 for partition monoTest-0
2025-12-04T16:12:40.123+09:00  INFO 10101 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : Record in retry and not yet recovered
... then no retry

Sample

A link to a GitHub repository with a minimal, reproducible, sample.

Reports that include a sample will take priority over reports that do not.
At times, we may require a sample, so it is good to try and include a sample up front.

[sobychacko]

@moonyoungCHAE It looks like a bug, didn't get a chance to triage deep. If you have ideas for a fix, please submit a PR. Thanks!

[manhnt27]

@moonyoungCHAE @sobychacko When I read this Stackoverflow and Github issue, I think we can break down into 2 ones here with async reply:

• Stackoverflow issue: Map mono error with exception --> Offset is immediately committed, no retry
• Github issue: Directly throw exception --> Offset not committed, retry only once, although configured more than 1 retry

I've read the source code and made some debugs:

• Stackoveflow: The code does not reach the common error handler
• Github: We leverage async reply so ack mod is set to MANUAL and asyncAck will be used. This would lead to pause the consumer and may prevent the consumer from retrieving new messages. So we are "stuck" in the first retry.

The main difference here is directly throw exception would allow us to jump to the default error handler, while wrapping exception with Mono error would go to the ack flow. Hence, it does not include default handling error that we've defined before.

When I read the doc about async reply, it seems that must catch exception and wrap it with CompletableFuture or Mono object.
Spring Kafka doc:

If some exception occurs within the listener method that prevents creation of the async result object, you MUST catch that exception and return an appropriate return object that will cause the message to be ack or recover.

As I know from this statement, we can configure some custom logics for recovery but retry is not allowed here? (Sorry I'm new to project. Correct me if anything is wrong. Thanks!)

Ref Kafka async reply doc: https://docs.spring.io/spring-kafka/reference/kafka/receiving-messages/async-returns.html

[moonyoungCHAE]

@manhnt27

Hi, I also analyzed the code related to this issue, and I observed the same behavior that you described.

• Stackoveflow : When an error occurs in a Mono, it does not reach the container error handler and the offset is committed immediately.
• Github: When a regular exception is thrown, it does reach the error handler, and depending on the error handler configuration, it may seek to a previous offset. However, retry does not work and the consumer ends up being stuck. (The consumer keeps polling, but offsetsInThisBatch is not cleared and the record is not re-consumed.)
  • To be more specific, this does not occur only with reactive listeners.
    It occurs when manual acknowledgment is used (for reactive listeners, the default is manual ack) and seekAfterHandle is set to true (which is the default).

Regarding your questions:

• Stackoveflow : When a Mono completes with an error, the error handler and retry mechanism do not work for now. → I think it could be reasonable to discuss this as a new feature for reactive listeners.
• Github: When a regular exception occurs, retry does not work. → This is the main problem described in this issue. Although the documentation states that exceptions must be caught in reactive listeners, as @sobychacko mentioned, this behavior looks more like a bug to me. - In the success case, offsets are committed automatically, but only in the failure case users are forced to manually control commit behavior. Requiring every developer who uses reactive listeners to always write explicit try–catch blocks does not seem like a good design approach.

@sobychacko
Correct me if anything is wrong. :)

Could you help confirm whether fixing this issue in this way is the right approach?

I’ve been working on a fix for this issue, and I noticed that when manual acknowledgment is used, offsetsInThisBatch is not cleared when an error occurs, which causes the container to get stuck.

Because of that, I’ve been approaching this by clearing offsetsInThisBatch when an error occurs. This is still a work in progress and not fully completed yet.

For example, in KafkaMessageListenerContainer.java, inside invokeErrorHandler, I’m doing the following:

List<ConsumerRecord<?, ?>> records = new ArrayList<>();
List<ConsumerRecord<K, V>> record2 = new ArrayList<>();
records.add(cRecord);
while (iterator.hasNext()) {
    ConsumerRecord<K, V> next = iterator.next();
    records.add(next);
    record2.add(next);
}
if (this.offsetsInThisBatch != null) {
    removeOffsetsInBatch(cRecord);
    for (ConsumerRecord<K, V> record : record2) {
        removeOffsetsInBatch(record);
    }
}

this.commonErrorHandler.handleRemaining(rte, records, this.consumer,
        KafkaMessageListenerContainer.this.thisOrParentContainer);

and the removeOffsetsInBatch implementation looks like this:

private synchronized void removeOffsetsInBatch(ConsumerRecord<K, V> cRecord) {
    if (this.offsetsInThisBatch == null || this.deferredOffsets == null) {
        return;
    }

    TopicPartition part = new TopicPartition(cRecord.topic(), cRecord.partition());
    List<Long> offs = this.offsetsInThisBatch.getOrDefault(part, new ArrayList<>());
    offs.remove(cRecord.offset());
    if (offs.isEmpty()) {
        this.deferredOffsets.remove(part);
        this.offsetsInThisBatch.remove(part);
        return;
    } else {
        this.offsetsInThisBatch.put(part, offs);
    }

    List<ConsumerRecord<K, V>> deferred =
            this.deferredOffsets.getOrDefault(part, new ArrayList<>());
    deferred.remove(cRecord);
    if (deferred.isEmpty()) {
        this.deferredOffsets.remove(part);
        this.offsetsInThisBatch.remove(part);
    } else {
        this.deferredOffsets.put(part, deferred);
    }
}

Previously, offsetsInThisBatch was only managed during commits.
With this change, I’m trying to also handle the case where an error occurs, so that the offsets do not remain stuck.

I’d appreciate your feedback on whether this direction makes sense.