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 messages to {topic}-dlt
        var recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate,
            (record, ex) -> new TopicPartition(record.topic() + "-dlt", record.partition()));

        // Retry 3 times with exponential backoff, then DLT
        var errorHandler = new DefaultErrorHandler(
            recoverer,
            new ExponentialBackOff(1000L, 2.0) {{
                setMaxElapsedTime(30_000L); // Max 30s total
            }}
        );

        // Don't retry these exceptions - go directly to DLT
        errorHandler.addNotRetryableExceptions(
            InvalidMessageException.class,
            JsonParseException.class
        );

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