Resolves #2009

The expected KafkaBackoffException's message is being logged at WARN level.

Make KafkaBackoffException be logged in DEBUG level

Remove coupling between KafkaMessageListenerContainer and KafkaBackoffException

Author: tomazfernandes

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -2495,12 +2495,7 @@ private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> recor
 					commitOffsetsIfNeeded(record);
 				}
 				catch (KafkaException ke) {
-					if (ke.contains(KafkaBackoffException.class)) {
-						this.logger.debug(ke.getMessage());
-					}
-					else {
-						ke.selfLog(ERROR_HANDLER_THREW_AN_EXCEPTION, this.logger);
-					}
+					ke.selfLog(ERROR_HANDLER_THREW_AN_EXCEPTION, this.logger);
 					return ke;
 				}
 				catch (RuntimeException ee) {

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -28,6 +28,7 @@
 
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.core.log.LogAccessor;
+import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.listener.AcknowledgingConsumerAwareMessageListener;
 import org.springframework.kafka.listener.CommonErrorHandler;
@@ -143,6 +144,7 @@ public void setErrorHandlerCustomizer(Consumer<CommonErrorHandler> errorHandlerC
 	private CommonErrorHandler createErrorHandler(DeadLetterPublishingRecoverer deadLetterPublishingRecoverer) {
 		DefaultErrorHandler errorHandler = new DefaultErrorHandler(deadLetterPublishingRecoverer,
 				new FixedBackOff(0, 0));
+		errorHandler.setLogLevel(KafkaException.Level.DEBUG);
 		errorHandler.setCommitRecovered(true);
 		this.errorHandlerCustomizer.accept(errorHandler);
 		return errorHandler;