How to seek & commit when using custom (ConsumerAware) error handler?

Question

Situation

I am trying to implement my custom error handler for a Spring-Kafka(2.3.3.RELEASE) batch message listener. I have pretty bare bone configuration:

val factory = ConcurrentKafkaListenerContainerFactory<Int, String>()

factory.consumerFactory = kafkaConsumerFactory
factory.isBatchListener = true
factory.setBatchErrorHandler(MyErrorHandler())

Without any extra consumer configuration. Note that "Starting with version 2.3, the framework sets enable.auto.commit to false".

I am using a "consumer aware" @KafkaListener and my custom error handler also implement ConsumerAwareBatchErrorHandler.

In my error handler I use the Consumer.seek(topicPartition, offset) method to either seek to next or seek to current offset.

The problem is that whenever the error handler is invoked and I manually seek to some offset, the offset doesn't seem to be committed to Kafka. This is proven by the fact that after an application restart it polls the same (previously failed) records (even when the code seeks to next). What confuses me even more is that while the application is running, it does seem to use the seek()'ed offset and polls new records.

I've tried using different AckModes, for example MANUAL or MANUAL_IMMIDIATE, but that does not change the observed behavior. Also the documentation says for these modes:

MANUAL, and MANUAL_IMMEDIATE require the listener to be an AcknowledgingMessageListener or a BatchAcknowledgingMessageListener

which obviously I am not using.

If have already found a "fix", which is calling consumer.commitSync() after the .seek(), which seems to commit the offset immediately. This is also backed up by the logging:

DEBUG 13784 --- [ntainer#2-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-1, groupId=kafka-retry-test] Committed offset 28 for partition test-retry-batch-custom-e-0

Question

Why is just calling consumer.seek(...) not committing offsets to kafka and am I missing some critical configuration? Calling the .commitSync() seems like a hack and is not described in the documentation to be needed anywhere.

Gary Russell Gary Russell · Accepted Answer · 2019-12-06T14:07:35

A seek does not commit the offset, it simply positions the current consumer at that point. You need to call syncCommit or asyncCommit.

However, starting with versionn 2.3.2 (current release is 2.3.4), you can return true from isAckAfterHandle():

/**
 * Return true if the offset should be committed for a handled error (no exception
 * thrown).
 * @return true to commit.
 * @since 2.3.2
 */
default boolean isAckAfterHandle() {
    // TODO: Default true in the next release.
    return false;
}

and the container will do the commits for you.

See the documentation.

Starting with version 2.3.2, these interfaces have a default method isAckAfterHandle() which is called by the container to determine whether the offset(s) should be committed if the error handler returns without throwing an exception. This returns false by default, for backwards compatibility. In most cases, however, we expect that the offset should be committed. For example, the SeekToCurrentErrorHandler returns true if a record is recovered (after any retries, if so configured). In a future release, we expect to change this default to true.

However, if you are using manual acks, your error handler must manually commit.

How to seek & commit when using custom (ConsumerAware) error handler?

1 Answers