spring kafka error handling without KafkaListener

I am using the spring kafka in my project and have implemented a retry functionality. A message is consumed and if it fails I need to retry. Below code is working fine if I am using and doing retry for some exceptions.

@Component
public class Consumer implements AcknowledgingMessageListener<String, String> {
@Override
@KafkaListener(topics = { "ff808081672c17c8016730733d020001.CBS_PROFILE"})
public void onMessage(ConsumerRecord<String, String> consumerRecord, Acknowledgment acknowledgment) {
log.info("Computed ");
throw new RuntimeException();
}
}
@Configuration
@Slf4j
public class KafkaConsumerConfig {
/** handles error and recovery logic **/
public DefaultErrorHandler errorHandler() {
System.out.println("Inside error handler");
// if these exceptions occur in the consumer, record will not be retried
List<Class<? extends Exception>> exceptionsToIgnore = List.of(JsonSchemaValidationException.class);
// if these exceptions occur in the consumer, record will be retried
List<Class<? extends Exception>> exceptionsToRetry = List.of(DataEngineNotAvailableException.class, ListenerExecutionFailedException.class);
// retry twice after fixed interval of one second
var fixedBackOff = new FixedBackOff(500L, 1);
var expBackoff = new ExponentialBackOffWithMaxRetries(2);
expBackoff.setInitialInterval(1000L);
var errorHandler = new DefaultErrorHandler(fixedBackOff);
errorHandler.setRetryListeners((consumerRecord, ex, deliveryAttempt) -> {
log.info("Failed record {} in retry listeners, Exception : {}, delivery attempt : {}", consumerRecord,
ex.getMessage(), deliveryAttempt);
});
exceptionsToIgnore.forEach(errorHandler::addNotRetryableExceptions);
exceptionsToRetry.forEach(errorHandler::addRetryableExceptions);
return errorHandler;
}
@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
ConsumerFactory<Object, Object> kafkaConsumerFactory) {
ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
configurer.configure(factory, kafkaConsumerFactory);
factory.getContainerProperties().setAckMode(AckMode.MANUAL);
// should be equal to number of partitions
factory.setConcurrency(3);
factory.setCommonErrorHandler(errorHandler());
return factory;
}
}

But I am not able to implement retry mechanism if I implement consumer like below.
Expectation is to retry the dataProcessor.readRecord(r, consumer) method.

@PostConstruct
public void onLoad() {
configService.setConfigurations();
//load consumer
consumerRunnerV2.loadConsumers();
}
public void loadConsumers() {
consumerServiceV2.fillConsumerMap();
Map<String, LRConsumer> consumersMap = consumerServiceV2.getConsumersMap();
for (Map.Entry<String, LRConsumer> consumer : consumersMap.entrySet()) {
consumerTaskExecutor.execute(() -> runConsumer(consumer.getValue()));
}
}
private void runConsumer(LRConsumer consumer) {
while (true) {
// to check state of consumer
consumer = consumerServiceV2.getConsumerById(consumer.getConsumerId());
if (consumer.getState() != ConsumerState.PAUSED) {
log.info("Waiting to receive messages.");
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> r : records)
//expectation is to retry this method
dataProcessor.readRecord(r, consumer);
}
else {
Thread.sleep(1000);
log.info("Consumer is in paused state.");
}
}
}

>ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

Of course it won't work with that; you are completely bypassing the Spring code and interacting with the KafkaConsumer yourself.

In that case you will have to implement your own error handling.

However, you don't have to use @KafkaListener to use a container, you can use the container factory to create a container and use setMessageListener to add your listener to the container properties.

You should remove the @KafkaListener in that case, so the framework doesn't automatically configure a container for it.