DefaultErrorHandler with Dead Letter Topic

@Configuration
@Slf4j
public class KafkaConsumerConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory(
        ConsumerFactory<String, Object> consumerFactory,
        KafkaTemplate<String, Object> kafkaTemplate
    ) {
        var factory = new ConcurrentKafkaListenerContainerFactory<String, Object>();
        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);

        // DeadLetterPublishingRecoverer sends failed records to {topic}-dlt, same partition
        var recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate,
            (record, ex) -> new TopicPartition(record.topic() + "-dlt", record.partition()));

        // 3 retries with exponential back off (1 s, 2 s, 4 s, capped at 10 s), then the recoverer
        var backOff = new ExponentialBackOffWithMaxRetries(3);
        backOff.setInitialInterval(1_000L);
        backOff.setMultiplier(2.0);
        backOff.setMaxInterval(10_000L);

        var errorHandler = new DefaultErrorHandler(recoverer, backOff);

        // Add exception types that must not be retried: they go straight to the recoverer
        errorHandler.addNotRetryableExceptions(IllegalArgumentException.class);

        // With MANUAL_IMMEDIATE, commit the offset of a record the recoverer handled
        errorHandler.setCommitRecovered(true);

        factory.setCommonErrorHandler(errorHandler);
        factory.setConcurrency(3);
        return factory;
    }
}