How to handle exception while using spring kafka asyncAck

[subham611]

I have a requirement to process kafka messages in at-least-once fashion. Spring kafka supports async ack starting from 2.8 version. I am storing received offsets from kafka in a map and after message processing is done committing kafka offsets. This all working fine until i send any error event(poison pill). I am not able to commit bad record inside error handler and due to this kafka is not consuming any new records after encountering any bad/malformed record.

code for kafka listener factory:

public ConcurrentKafkaListenerContainerFactory<String, JsonNode> kafkaListenerContainerFactory(ConsumerFactory<String, JsonNode> kafkaConsumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, JsonNode> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(kafkaConsumerFactory);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        factory.setErrorHandler(errorHandler());
        factory.getContainerProperties().setAsyncAcks(true);
        return factory;
    }

Error Handler Code:

@Bean("errorHandler")
    public ErrorHandler errorHandler() {
        log.info("Creating error handler");
        return (thrownException, records) -> {
            log.error("Inside error handler");
        };
    }

ErrorHandler is marked as Deprecated. Even in CommonErrorHandler I am not able to overcome this issue.

[garyrussell]

Please don't ask the same question in multiple places; it is a waste of your time and ours.

See my answer on Stack Overflow https://stackoverflow.com/questions/75172594/how-to-handle-exception-while-using-spring-kafka-asyncack/75173726#75173726

My original comment about MANUAL_IMMEDIATE there was invalid; it is not available with async acks.