Provide ability to disable/limit retries in batch listener when using DefaultAfterRollbackProcessor

[RuslanHryn]

Expected Behavior
Configuration to disable/limit retries in the batch listener when using DB transactions and DefaultAfterRollbackProcessor
Example of listener @KafkaListener(topics = BATCH_TOPIC, batch = "true") {}

Current Behavior
Currently, the DefaultAfterRollbackProcessor retries the failing message infinitely if using batch listener with DB transactions.

Context
The main problem if we forget to handle the failing message inside the listener then the failed message will stuck until we take some manual action to skip this message. That's why we would like to have a config to disable/limit retries at the configurations level.
I tried to find a way to disable retries for batch listeners but unsuccessfully.
new FixedBackOff(0, 0)
new DefaultAfterRollbackProcessor<>(deadLetterPublishingRecoverer, backOff);

The workaround can be to force to use recoverable=true always in order to limit retries

return new DefaultAfterRollbackProcessor<>(deadLetterPublishingRecoverer, backOff) {
            @Override
            public void process(
                    @NotNull List<ConsumerRecord<Object, Object>> consumerRecords,
                    @NotNull org.apache.kafka.clients.consumer.Consumer<Object, Object> consumer,
                    MessageListenerContainer container,
                    @NotNull Exception exception,
                    boolean recoverable,
                    @NotNull ContainerProperties.EOSMode eosMode) {
                // Force all messages to be recoverable.
                super.process(consumerRecords, consumer, container, exception, true, eosMode);
            }
        };

[garyrussell]

Forcing it to be recoverable will not do what you expect. We don't know which record in the batch failed, the whole batch is passed in and the "recovered" record will just be the first one in the batch and not necessarily the one that is causing the failure; that's why batch failures are not recoverable.

To solve this problem, we should probably enhance the DARP to understand BatchListenerFailedException s (like the DefaultErrorHandler now does) so the app can indicate which record caused the failure and only recover that one.

We could also provide recovery for other exceptions by recovering every record in the batch (again, like the DefaultErrorHandler).

Note that if you are only using DB transactions (and not publishing to Kafka as well as updating the DB in the listener), there is no need to make the container transactional.

[RuslanHryn]

Forcing it to be recoverable will not do what you expect. We don't know which record in the batch failed, the whole batch is passed in and the "recovered" record will just be the first one in the batch and not necessarily the one that is causing the failure; that's why batch failures are not recoverable.

Yes, you are right. It will just limit retries. It will not be infinite

Note that if you are only using DB transactions (and not publishing to Kafka as well as updating the DB in the listener), there is no need to make the container transactional.

we always update DB in RecordFilterStrategy to save processed messages into DB in order to implement an idempotent consumer pattern. That's why we can't use DefaultErrorHandler

[garyrussell]

You don’t need Kafka transactions if you are not producing to Kafka in your listener. They have nothing to do with idempotent consumers, only exactly once semantics for consume->process->produce scenarios.

Or, for atomically producing multiple records.

[RuslanHryn]

You don’t need Kafka transactions if you are not producing to Kafka in your listener. They have nothing to do with idempotent consumers, only exactly once semantics for consume->process->produce scenarios.

Or, for atomically producing multiple records.

We don't use Kafka transactions at all. We use DB transactions instead to implement idempotent consumer. Anyway, it does not matter which transactions to use there, the behavior of retries will be the same. So it would be nice to have proper retries in the batch listener with DefaultAfterRollbackProcessor or an option to disable retries to avoid the infinite loop

[garyrussell]

I agree; hence my comments above about enhancing the rollback processor.

However, adding a transaction manager to the container is only intended for the use of Kafka transactions.

You can use @Transactional on the listener method or, to include the listener adapter (including any RecordFilterStrategy) in the scope of the transaction, add a TransactionInterceptor to the container's adviceChain property.

The container should never have a transaction manager if Kafka transactions are not being used.

[Wzy19930507]

Hi, @garyrussell may i pick it up?

[artembilan]

@Wzy19930507 ,

sure you can!

This is an Open Source project, so any contribution is welcome!
I cannot advice anything for you what exactly to do at the moment, but let's see first if you have anything in mind how to fix it.
Don't hesitate to open Pull Request: https://github.com/spring-projects/spring-kafka/blob/main/CONTRIBUTING.adoc !