Kafka Batch Consumer - Offset commit handling

Question

I am using spring-kafka 2.2.8 and trying to create a batch consumer. Here i've a question.

Let's say I'm receiving around 100 records in each poll based on my consumer settings and out of these 100 records, if only one record is failing because of some downstream system error ( it could be a transient error like DB is down for couple of seconds and came back etc), then I'm trying to understand if I can commit the offset for all those 100 records and send the failed record to a different topic or some other persistent store to handle it later?

Answer 1

Starting with version 2.5, there is now a RecoveringBatchErrorHandler; with this error handler, you throw a "special" exception to tell the error handler which record in the batch failed. The error handler will commit the previous offset(s) and re-seek the failed record.

You can add a back-off to delay the redelivery for a short while. When retries are exhausted, the record is "recovered" (e.g. by sending to a dead letter topic).

Documentation here.

As an alternative to the Retrying Batch Error Handler, version 2.5 introduced the RecoveringBatchErrorHandler.

This is now the default error handler for batch listeners. The default configuration retries 9 times (10 delivery attempts) with no back off between deliveries.

This error handler works in conjunction with the listener throwing a BatchListenerFailedException providing the index in the batch where the failure occurred. If the listener throws a different exception, or the index is out of range, the error handler falls back to invoking a SeekToCurrentBatchErrorHandler and the whole batch is retried, with no recovery available. The sequence of events is:

Commit the offsets of the records before the index.

If retries are not exhausted, perform seeks so that all the remaining records (including the failed record) will be redelivered.

If retries are exhausted, attempt recovery of the failed record (default log only) and perform seeks so that the remaining records (excluding the failed record) will be redelivered. The recovered record’s offset is committed

If retries are exhausted and recovery fails, seeks are performed as if retries are not exhausted.

The default recoverer logs the failed record after retries are exhausted. You can use a custom recoverer, or one provided by the framework such as the DeadLetterPublishingRecoverer.

@Bean
public RecoveringBatchErrorHandler(KafkaTemplate<String, String> template) {
    DeadLetterPublishingRecoverer recoverer =
            new DeadLetterPublishingRecoverer(template);
    RecoveringBatchErrorHandler errorHandler =
            new RecoveringBatchErrorHandler(recoverer, new FixedBackOff(2L, 5000));
}

You can either provide the failed record, or its index in the exception:

@KafkaListener(id = "recovering", topics = "someTopic")
public void listen(List<ConsumerRecord<String, String>> records) {
    records.forEach(record -> {
        try {
            process(record);
        }
        catch (Exception e) {
            throw new BatchListenerFailedException("Failed to process", record);
        }
    });
}