BackoffExceptions are logged at error level when using RetryTopicConfiguration

Question

I am a happy user of the recently added RetryTopicConfiguration there is however a small issue that is bothering me. The setup I use looks like:

@Bean
public RetryTopicConfiguration retryTopicConfiguration(
    KafkaTemplate<String, String> template,
    @Value("${kafka.topic.in}") String topicToInclude,
    @Value("${spring.application.name}") String appName) {
    return RetryTopicConfigurationBuilder
        .newInstance()
        .fixedBackOff(5000L)
        .maxAttempts(3)
        .retryTopicSuffix("-" + appName + ".retry")
        .suffixTopicsWithIndexValues()
        .dltSuffix("-" + appName + ".dlq")
        .includeTopic(topicToInclude)
        .dltHandlerMethod(KAFKA_EVENT_LISTENER, "handleDltEvent")
        .create(template);
}

When the a listener throws an exception that triggers a retry, the DefaultErrorHandler will log a KafkaBackoffException at error level.

For a similar problem it was suggested to use a ListenerContainerFactoryConfigurer yet this does not remove all error logs, since I still see the following in my logs:

2022-04-02 17:34:33.340 ERROR 8054 --- [e.retry-0-0-C-1] o.s.kafka.listener.DefaultErrorHandler   : Recovery of record (topic-spring-kafka-logging-issue.retry-0-0@0) failed

org.springframework.kafka.listener.ListenerExecutionFailedException: Listener failed; nested exception is org.springframework.kafka.listener.KafkaBackoffException: Partition 0 from topic topic-spring-kafka-logging-issue.retry-0 is not ready for consumption, backing off for approx. 4468 millis.

Can the log-level be changed, without adding a custom ErrorHandler?

Spring-Boot version: 2.6.6
Spring-Kafka version: 2.8.4
JDK version: 11

Sample project: here

Answer 1

Thanks for such a complete question. This is a known issue of Spring for Apache Kafka 2.8.4 due to the new combine blocking and non-blocking exceptions feature and has been fixed for 2.8.5.

The workaround is to clear the blocking exceptions mechanism such as:

@Bean(name = RetryTopicInternalBeanNames.LISTENER_CONTAINER_FACTORY_CONFIGURER_NAME)
public ListenerContainerFactoryConfigurer lcfc(KafkaConsumerBackoffManager kafkaConsumerBackoffManager,
                        DeadLetterPublishingRecovererFactory deadLetterPublishingRecovererFactory,
                        @Qualifier(RetryTopicInternalBeanNames
                            .INTERNAL_BACKOFF_CLOCK_BEAN_NAME) Clock clock) {
    ListenerContainerFactoryConfigurer lcfc = new ListenerContainerFactoryConfigurer(kafkaConsumerBackoffManager, deadLetterPublishingRecovererFactory, clock);

    lcfc.setBlockingRetriesBackOff(new FixedBackOff(0, 0));
    lcfc.setErrorHandlerCustomizer(eh -> ((DefaultErrorHandler) eh).setClassifications(Collections.emptyMap(), true));
    return lcfc;
}

Please let me know if that works for you.

Thanks.

EDIT: This workaround disables only blocking retries, which since 2.8.4 can be used along non-blocking as per the link in the original answer. The exception classification for the non-blocking retries is in the DefaultDestinationTopicResolver class, and you can set FATAL exceptions as documented here.

EDIT: Alternatively, you can use the Spring Kafka 2.8.5-SNAPSHOT version by adding the Spring Snapshot repository such as:

repositories {
    maven {
        url 'https://repo.spring.io/snapshot'
    }
}

dependencies {
    implementation 'org.springframework.kafka:spring-kafka:2.8.5-SNAPSHOT'
}

You can also downgrade to Spring Kafka 2.8.3.

As Gary Russell pointed out, if your application is already in production you should not use the SNAPSHOT version, and 2.8.5 is out in a couple of weeks.

EDIT 2: Glad to hear you’re happy about the feature!