Use LogAccessor for logging

- use lambdas instead of testing logging level.
- add tracing to `CloseSafeProducer`

spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java

@@ -35,7 +35,6 @@ import java.util.stream.Collectors;
 
 import org.I0Itec.zkclient.ZkClient;
 import org.I0Itec.zkclient.exception.ZkInterruptedException;
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
@@ -50,6 +49,7 @@ import org.apache.kafka.common.utils.Time;
 
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.test.core.BrokerAddress;
 import org.springframework.retry.backoff.ExponentialBackOffPolicy;
 import org.springframework.retry.policy.SimpleRetryPolicy;
@@ -80,7 +80,7 @@ import kafka.zk.EmbeddedZookeeper;
  */
 public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean {
 
-	private static final Log logger = LogFactory.getLog(EmbeddedKafkaBroker.class); // NOSONAR
+	private static final LogAccessor logger = new LogAccessor(LogFactory.getLog(EmbeddedKafkaBroker.class)); // NOSONAR
 
 	public static final String BEAN_NAME = "embeddedKafka";
 
@@ -471,9 +471,7 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean {
 			@Override
 			public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 				assigned.set(true);
-				if (logger.isDebugEnabled()) {
-					logger.debug("partitions assigned: " + partitions);
-				}
+				logger.debug(() -> "partitions assigned: " + partitions);
 			}
 
 		});
@@ -484,14 +482,12 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean {
 		}
 		if (records != null && records.count() > 0) {
 			final ConsumerRecords<?, ?> theRecords = records;
-			if (logger.isDebugEnabled()) {
-				logger.debug("Records received on initial poll for assignment; re-seeking to beginning; "
-						+ records.partitions().stream()
-						.flatMap(p -> theRecords.records(p).stream())
-						// map to same format as send metadata toString()
-						.map(r -> r.topic() + "-" + r.partition() + "@" + r.offset())
-						.collect(Collectors.toList()));
-			}
+			logger.debug(() -> "Records received on initial poll for assignment; re-seeking to beginning; "
+					+ theRecords.partitions().stream()
+					.flatMap(p -> theRecords.records(p).stream())
+					// map to same format as send metadata toString()
+					.map(r -> r.topic() + "-" + r.partition() + "@" + r.offset())
+					.collect(Collectors.toList()));
 			consumer.seekToBeginning(records.partitions());
 		}
 		assertThat(assigned.get())

spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java

@@ -24,7 +24,6 @@ import java.util.Map;
 import java.util.Properties;
 import java.util.stream.Collectors;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -37,6 +36,7 @@ import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 
 import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.util.Assert;
 
@@ -49,7 +49,7 @@ import org.springframework.util.Assert;
  */
 public final class KafkaTestUtils {
 
-	private static final Log logger = LogFactory.getLog(KafkaTestUtils.class); // NOSONAR
+	private static final LogAccessor logger = new LogAccessor(LogFactory.getLog(KafkaTestUtils.class)); // NOSONAR
 
 	private static Properties defaults;
 
@@ -196,14 +196,12 @@ public final class KafkaTestUtils {
 	public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> consumer, long timeout) {
 		logger.debug("Polling...");
 		ConsumerRecords<K, V> received = consumer.poll(Duration.ofMillis(timeout));
-		if (logger.isDebugEnabled()) {
-			logger.debug("Received: " + received.count() + ", "
-					+ received.partitions().stream()
-					.flatMap(p -> received.records(p).stream())
-					// map to same format as send metadata toString()
-					.map(r -> r.topic() + "-" + r.partition() + "@" + r.offset())
-					.collect(Collectors.toList()));
-		}
+		logger.debug(() -> "Received: " + received.count() + ", "
+				+ received.partitions().stream()
+				.flatMap(p -> received.records(p).stream())
+				// map to same format as send metadata toString()
+				.map(r -> r.topic() + "-" + r.partition() + "@" + r.offset())
+				.collect(Collectors.toList()));
 		assertThat(received).as("null received from consumer.poll()").isNotNull();
 		return received;
 	}

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

@@ -35,7 +35,6 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.regex.Pattern;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.aop.framework.Advised;
@@ -62,6 +61,7 @@ import org.springframework.core.annotation.AnnotatedElementUtils;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.core.convert.converter.GenericConverter;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.format.Formatter;
 import org.springframework.format.FormatterRegistry;
 import org.springframework.format.support.DefaultFormattingConversionService;
@@ -138,7 +138,7 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 
 	private final Set<Class<?>> nonAnnotatedClasses = Collections.newSetFromMap(new ConcurrentHashMap<>(64));
 
-	private final Log logger = LogFactory.getLog(getClass());
+	private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass()));
 
 	private final ListenerScope listenerScope = new ListenerScope();
 
@@ -290,9 +290,7 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 			}
 			if (annotatedMethods.isEmpty()) {
 				this.nonAnnotatedClasses.add(bean.getClass());
-				if (this.logger.isTraceEnabled()) {
-					this.logger.trace("No @KafkaListener annotations found on bean type: " + bean.getClass());
-				}
+				this.logger.trace(() -> "No @KafkaListener annotations found on bean type: " + bean.getClass());
 			}
 			else {
 				// Non-empty set of methods
@@ -302,10 +300,8 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 						processKafkaListener(listener, method, bean, beanName);
 					}
 				}
-				if (this.logger.isDebugEnabled()) {
-					this.logger.debug(annotatedMethods.size() + " @KafkaListener methods processed on bean '"
+				this.logger.debug(() -> annotatedMethods.size() + " @KafkaListener methods processed on bean '"
 							+ beanName + "': " + annotatedMethods);
-				}
 			}
 			if (hasClassLevelListeners) {
 				processMultiMethodListeners(classLevelListeners, multiMethods, bean, beanName);
@@ -476,7 +472,7 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 						properties.load(new StringReader(value));
 					}
 					catch (IOException e) {
-						this.logger.error("Failed to load property " + property + ", continuing...", e);
+						this.logger.error(e, () -> "Failed to load property " + property + ", continuing...");
 					}
 				}
 			}

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java

@@ -21,13 +21,13 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.regex.Pattern;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
@@ -62,7 +62,7 @@ import org.springframework.util.Assert;
 public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMessageListenerContainer<K, V>, K, V>
 		implements KafkaListenerContainerFactory<C>, ApplicationEventPublisherAware, InitializingBean {
 
-	protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR protected
+	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); // NOSONAR protected
 
 	private final ContainerProperties containerProperties = new ContainerProperties((Pattern) null);
 

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java

@@ -23,7 +23,6 @@ import java.util.Collections;
 import java.util.Properties;
 import java.util.regex.Pattern;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.BeansException;
@@ -34,6 +33,7 @@ import org.springframework.beans.factory.config.BeanExpressionContext;
 import org.springframework.beans.factory.config.BeanExpressionResolver;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.context.expression.BeanFactoryResolver;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.expression.BeanResolver;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.listener.BatchMessageListener;
@@ -67,7 +67,7 @@ import org.springframework.util.Assert;
 public abstract class AbstractKafkaListenerEndpoint<K, V>
 		implements KafkaListenerEndpoint, BeanFactoryAware, InitializingBean {
 
-	private final Log logger = LogFactory.getLog(getClass());
+	private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass()));
 
 	private String id;
 
@@ -458,10 +458,8 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 		if (this.recordFilterStrategy != null) {
 			if (this.batchListener) {
 				if (((MessagingMessageListenerAdapter<K, V>) messageListener).isConsumerRecords()) {
-					if (this.logger.isWarnEnabled()) {
-						this.logger.warn("Filter strategy ignored when consuming 'ConsumerRecords'"
+					this.logger.warn(() -> "Filter strategy ignored when consuming 'ConsumerRecords'"
 								+ (this.id != null ? " id: " + this.id : ""));
-					}
 				}
 				else {
 					messageListener = new FilteringBatchMessageListenerAdapter<>(

spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java

@@ -25,7 +25,6 @@ import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.BeansException;
@@ -38,6 +37,7 @@ import org.springframework.context.ApplicationListener;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.context.event.ContextRefreshedEvent;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
 import org.springframework.kafka.listener.MessageListenerContainer;
 import org.springframework.util.Assert;
@@ -68,7 +68,7 @@ import org.springframework.util.StringUtils;
 public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifecycle, ApplicationContextAware,
 		ApplicationListener<ContextRefreshedEvent> {
 
-	protected final Log logger = LogFactory.getLog(getClass()); //NOSONAR
+	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); //NOSONAR
 
 	private final Map<String, MessageListenerContainer> listenerContainers =
 			new ConcurrentHashMap<String, MessageListenerContainer>();
@@ -230,7 +230,7 @@ public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifec
 					((DisposableBean) listenerContainer).destroy();
 				}
 				catch (Exception ex) {
-					this.logger.warn("Failed to destroy message listener container", ex);
+					this.logger.warn(ex, "Failed to destroy message listener container");
 				}
 			}
 		}

spring-kafka/src/main/java/org/springframework/kafka/config/MethodKafkaListenerEndpoint.java

@@ -19,11 +19,11 @@ package org.springframework.kafka.config;
 import java.lang.reflect.Method;
 import java.util.Arrays;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.listener.KafkaListenerErrorHandler;
 import org.springframework.kafka.listener.MessageListenerContainer;
 import org.springframework.kafka.listener.adapter.BatchMessagingMessageListenerAdapter;
@@ -53,7 +53,7 @@ import org.springframework.util.Assert;
  */
 public class MethodKafkaListenerEndpoint<K, V> extends AbstractKafkaListenerEndpoint<K, V> {
 
-	private final Log logger = LogFactory.getLog(getClass());
+	private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass()));
 
 	private Object bean;
 
@@ -112,12 +112,11 @@ public class MethodKafkaListenerEndpoint<K, V> extends AbstractKafkaListenerEndp
 		if (replyingMethod != null) {
 			SendTo ann = AnnotationUtils.getAnnotation(replyingMethod, SendTo.class);
 			if (ann != null) {
-				if (replyingMethod.getReturnType().equals(void.class)
-					&& this.logger.isWarnEnabled()) {
-						this.logger.warn("Method "
-								+ replyingMethod
-								+ " has a void return type; @SendTo is ignored" +
-								(this.errorHandler == null ? "" : " unless the error handler returns a result"));
+				if (replyingMethod.getReturnType().equals(void.class)) {
+					this.logger.warn(() -> "Method "
+							+ replyingMethod
+							+ " has a void return type; @SendTo is ignored" +
+							(this.errorHandler == null ? "" : " unless the error handler returns a result"));
 				}
 				String[] destinations = ann.value();
 				if (destinations.length > 1) {

spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java

@@ -20,7 +20,6 @@ import java.time.Duration;
 import java.util.Map;
 import java.util.Properties;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.streams.KafkaClientSupplier;
 import org.apache.kafka.streams.KafkaStreams;
@@ -32,6 +31,7 @@ import org.apache.kafka.streams.processor.internals.DefaultKafkaClientSupplier;
 
 import org.springframework.beans.factory.config.AbstractFactoryBean;
 import org.springframework.context.SmartLifecycle;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.core.CleanupConfig;
 import org.springframework.lang.Nullable;
@@ -60,7 +60,7 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 	 */
 	public static final Duration DEFAULT_CLOSE_TIMEOUT = Duration.ofSeconds(10);
 
-	private static final Log LOGGER = LogFactory.getLog(StreamsBuilderFactoryBean.class);
+	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(StreamsBuilderFactoryBean.class));
 
 	private static final String STREAMS_CONFIG_MUST_NOT_BE_NULL = "'streamsConfig' must not be null";
 
@@ -291,9 +291,7 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 				Assert.state(this.streamsConfig != null || this.properties != null,
 						"'streamsConfig' or streams configuration properties must not be null");
 				Topology topology = getObject().build(); // NOSONAR
-				if (LOGGER.isDebugEnabled()) {
-					LOGGER.debug(topology.describe());
-				}
+				LOGGER.debug(() -> topology.describe().toString());
 				if (this.properties != null) {
 					this.kafkaStreams = new KafkaStreams(topology, this.properties, this.clientSupplier);
 				}
@@ -331,7 +329,7 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 				}
 			}
 			catch (Exception e) {
-				LOGGER.error("Failed to stop streams", e);
+				LOGGER.error(e, "Failed to stop streams");
 			}
 			finally {
 				this.running = false;

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

@@ -30,7 +30,6 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.producer.Callback;
@@ -52,6 +51,7 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.ApplicationListener;
 import org.springframework.context.event.ContextStoppedEvent;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.support.TransactionSupport;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -93,7 +93,7 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 	 */
 	public static final Duration DEFAULT_PHYSICAL_CLOSE_TIMEOUT = Duration.ofSeconds(30);
 
-	private static final Log LOGGER = LogFactory.getLog(DefaultKafkaProducerFactory.class);
+	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(DefaultKafkaProducerFactory.class));
 
 	private final Map<String, Object> configs;
 
@@ -145,11 +145,9 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 		String txId = (String) this.configs.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG);
 		if (StringUtils.hasText(txId)) {
 			setTransactionIdPrefix(txId);
-			if (LOGGER.isInfoEnabled()) {
-				LOGGER.info("If 'setTransactionIdPrefix()' is not going to be configured, " +
-						"an existing 'transactional.id' config with value: '" + txId +
-						"' will be suffixed with the number for concurrent transactions support.");
-			}
+			LOGGER.info(() -> "If 'setTransactionIdPrefix()' is not going to be configured, "
+					+ "the existing 'transactional.id' config with value: '" + txId
+					+ "' will be suffixed for concurrent transactions support.");
 		}
 	}
 
@@ -198,9 +196,9 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 	 */
 	private void enableIdempotentBehaviour() {
 		Object previousValue = this.configs.putIfAbsent(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
-		if (LOGGER.isDebugEnabled() && Boolean.FALSE.equals(previousValue)) {
-			LOGGER.debug("The '" + ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG +
-					"' is set to false, may result in duplicate messages");
+		if (Boolean.FALSE.equals(previousValue)) {
+			LOGGER.debug(() -> "The '" + ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG
+					+ "' is set to false, may result in duplicate messages");
 		}
 	}
 
@@ -254,7 +252,7 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 				producerToClose.delegate.close(this.physicalCloseTimeout);
 			}
 			catch (Exception e) {
-				LOGGER.error("Exception while closing producer", e);
+				LOGGER.error(e, "Exception while closing producer");
 			}
 			producerToClose = this.cache.poll();
 		}
@@ -278,7 +276,22 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 	 * @since 2.2
 	 */
 	public void reset() {
-		destroy();
+		try {
+			destroy();
+		}
+		catch (Exception e) {
+			LOGGER.error(e, "Exception while closing producer");
+		}
+	}
+
+	/**
+	 * NoOp.
+	 * @return always true.
+	 * @deprecated {@link org.springframework.context.Lifecycle} is no longer implemented.
+	 */
+	@Deprecated
+	public boolean isRunning() {
+		return true;
 	}
 
 	@Override
@@ -430,16 +443,19 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 
 		@Override
 		public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
+			LOGGER.trace(() -> toString() + " send(" + record + ")");
 			return this.delegate.send(record);
 		}
 
 		@Override
 		public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
+			LOGGER.trace(() -> toString() + " send(" + record + ")");
 			return this.delegate.send(record, callback);
 		}
 
 		@Override
 		public void flush() {
+			LOGGER.trace(() -> toString() + " flush()");
 			this.delegate.flush();
 		}
 
@@ -460,16 +476,12 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 
 		@Override
 		public void beginTransaction() throws ProducerFencedException {
-			if (LOGGER.isDebugEnabled()) {
-				LOGGER.debug("beginTransaction: " + this);
-			}
+			LOGGER.debug(() -> toString() + " beginTransaction()");
 			try {
 				this.delegate.beginTransaction();
 			}
 			catch (RuntimeException e) {
-				if (LOGGER.isErrorEnabled()) {
-					LOGGER.error("beginTransaction failed: " + this, e);
-				}
+				LOGGER.error(e, () -> "beginTransaction failed: " + this);
 				this.txFailed = true;
 				throw e;
 			}
@@ -479,21 +491,18 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 		public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId)
 				throws ProducerFencedException {
 
+			LOGGER.trace(() -> toString() + " sendOffsetsToTransaction(" + offsets + ", " + consumerGroupId + ")");
 			this.delegate.sendOffsetsToTransaction(offsets, consumerGroupId);
 		}
 
 		@Override
 		public void commitTransaction() throws ProducerFencedException {
-			if (LOGGER.isDebugEnabled()) {
-				LOGGER.debug("commitTransaction: " + this);
-			}
+			LOGGER.debug(() -> toString() + " commitTransaction()");
 			try {
 				this.delegate.commitTransaction();
 			}
 			catch (RuntimeException e) {
-				if (LOGGER.isErrorEnabled()) {
-					LOGGER.error("commitTransaction failed: " + this, e);
-				}
+				LOGGER.error(e, () -> "commitTransaction failed: " + this);
 				this.txFailed = true;
 				throw e;
 			}
@@ -501,16 +510,12 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 
 		@Override
 		public void abortTransaction() throws ProducerFencedException {
-			if (LOGGER.isDebugEnabled()) {
-				LOGGER.debug("abortTransaction: " + this);
-			}
+			LOGGER.debug(() -> toString() + " abortTransaction()");
 			try {
 				this.delegate.abortTransaction();
 			}
 			catch (RuntimeException e) {
-				if (LOGGER.isErrorEnabled()) {
-					LOGGER.error("Abort failed: " + this, e);
-				}
+				LOGGER.error(e, () -> "Abort failed: " + this);
 				this.txFailed = true;
 				throw e;
 			}
@@ -530,13 +535,12 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 
 		@Override
 		public void close(@Nullable Duration timeout) {
+			LOGGER.trace(() -> toString() + " close(" + (timeout == null ? "null" : timeout) + ")");
 			if (this.cache != null) {
 				if (this.txFailed) {
-					if (LOGGER.isWarnEnabled()) {
-						LOGGER.warn("Error during transactional operation; producer removed from cache; possible " +
-								"cause: "
-								+ "broker restarted during transaction: " + this);
-					}
+					LOGGER.warn(() -> "Error during transactional operation; producer removed from cache; "
+							+ "possible cause: "
+							+ "broker restarted during transaction: " + this);
 					if (timeout == null) {
 						this.delegate.close();
 					}

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java

@@ -28,7 +28,6 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.stream.Collectors;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.CreatePartitionsResult;
@@ -45,6 +44,7 @@ import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.SmartInitializingSingleton;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.KafkaException;
 
 /**
@@ -65,7 +65,7 @@ public class KafkaAdmin implements ApplicationContextAware, SmartInitializingSin
 
 	private static final int DEFAULT_OPERATION_TIMEOUT = 30;
 
-	private static final Log LOGGER = LogFactory.getLog(KafkaAdmin.class);
+	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(KafkaAdmin.class));
 
 	private final Map<String, Object> config;
 
@@ -166,7 +166,7 @@ public class KafkaAdmin implements ApplicationContextAware, SmartInitializingSin
 					throw new IllegalStateException("Could not create admin", e);
 				}
 				else {
-					LOGGER.error("Could not create admin", e);
+					LOGGER.error(e, "Could not create admin");
 				}
 			}
 			if (adminClient != null) {
@@ -179,7 +179,7 @@ public class KafkaAdmin implements ApplicationContextAware, SmartInitializingSin
 						throw new IllegalStateException("Could not configure topics", e);
 					}
 					else {
-						LOGGER.error("Could not configure topics", e);
+						LOGGER.error(e, "Could not configure topics");
 					}
 				}
 				finally {
@@ -220,19 +220,15 @@ public class KafkaAdmin implements ApplicationContextAware, SmartInitializingSin
 			try {
 				TopicDescription topicDescription = f.get(this.operationTimeout, TimeUnit.SECONDS);
 				if (topic.numPartitions() < topicDescription.partitions().size()) {
-					if (LOGGER.isInfoEnabled()) {
-						LOGGER.info(String.format(
-							"Topic '%s' exists but has a different partition count: %d not %d", n,
-							topicDescription.partitions().size(), topic.numPartitions()));
-					}
+					LOGGER.info(() -> String.format(
+						"Topic '%s' exists but has a different partition count: %d not %d", n,
+						topicDescription.partitions().size(), topic.numPartitions()));
 				}
 				else if (topic.numPartitions() > topicDescription.partitions().size()) {
-					if (LOGGER.isInfoEnabled()) {
-						LOGGER.info(String.format(
-							"Topic '%s' exists but has a different partition count: %d not %d, increasing "
-							+ "if the broker supports it", n,
-							topicDescription.partitions().size(), topic.numPartitions()));
-					}
+					LOGGER.info(() -> String.format(
+						"Topic '%s' exists but has a different partition count: %d not %d, increasing "
+						+ "if the broker supports it", n,
+						topicDescription.partitions().size(), topic.numPartitions()));
 					topicsToModify.put(n, NewPartitions.increaseTo(topic.numPartitions()));
 				}
 			}
@@ -256,17 +252,17 @@ public class KafkaAdmin implements ApplicationContextAware, SmartInitializingSin
 		}
 		catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
-			LOGGER.error("Interrupted while waiting for topic creation results", e);
+			LOGGER.error(e, "Interrupted while waiting for topic creation results");
 		}
 		catch (TimeoutException e) {
 			throw new KafkaException("Timed out waiting for create topics results", e);
 		}
 		catch (ExecutionException e) {
 			if (e.getCause() instanceof TopicExistsException) { // Possible race with another app instance
-				LOGGER.debug("Failed to create topics", e.getCause());
+				LOGGER.debug(e.getCause(), "Failed to create topics");
 			}
 			else {
-				LOGGER.error("Failed to create topics", e.getCause());
+				LOGGER.error(e.getCause(), "Failed to create topics");
 				throw new KafkaException("Failed to create topics", e.getCause()); // NOSONAR
 			}
 		}
@@ -279,17 +275,17 @@ public class KafkaAdmin implements ApplicationContextAware, SmartInitializingSin
 		}
 		catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
-			LOGGER.error("Interrupted while waiting for partition creation results", e);
+			LOGGER.error(e, "Interrupted while waiting for partition creation results");
 		}
 		catch (TimeoutException e) {
 			throw new KafkaException("Timed out waiting for create partitions results", e);
 		}
 		catch (ExecutionException e) {
 			if (e.getCause() instanceof InvalidPartitionsException) { // Possible race with another app instance
-				LOGGER.debug("Failed to create partitions", e.getCause());
+				LOGGER.debug(e.getCause(), "Failed to create partitions");
 			}
 			else {
-				LOGGER.error("Failed to create partitions", e.getCause());
+				LOGGER.error(e.getCause(), "Failed to create partitions");
 				if (!(e.getCause() instanceof UnsupportedVersionException)) {
 					throw new KafkaException("Failed to create partitions", e.getCause()); // NOSONAR
 				}

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -19,7 +19,6 @@ package org.springframework.kafka.core;
 import java.util.List;
 import java.util.Map;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.producer.Callback;
@@ -30,6 +29,7 @@ import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.KafkaUtils;
 import org.springframework.kafka.support.LoggingProducerListener;
@@ -62,7 +62,7 @@ import org.springframework.util.concurrent.SettableListenableFuture;
  */
 public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
 
-	protected final Log logger = LogFactory.getLog(this.getClass()); //NOSONAR
+	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass())); //NOSONAR
 
 	private final ProducerFactory<K, V> producerFactory;
 
@@ -370,17 +370,13 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
 						+ "run in a transaction started by a listener container when consuming a record");
 		}
 		final Producer<K, V> producer = getTheProducer();
-		if (this.logger.isTraceEnabled()) {
-			this.logger.trace("Sending: " + producerRecord);
-		}
+		this.logger.trace(() -> "Sending: " + producerRecord);
 		final SettableListenableFuture<SendResult<K, V>> future = new SettableListenableFuture<>();
 		producer.send(producerRecord, buildCallback(producerRecord, producer, future));
 		if (this.autoFlush) {
 			flush();
 		}
-		if (this.logger.isTraceEnabled()) {
-			this.logger.trace("Sent: " + producerRecord);
-		}
+		this.logger.trace(() -> "Sent: " + producerRecord);
 		return future;
 	}
 
@@ -393,18 +389,14 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
 					if (KafkaTemplate.this.producerListener != null) {
 						KafkaTemplate.this.producerListener.onSuccess(producerRecord, metadata);
 					}
-					if (KafkaTemplate.this.logger.isTraceEnabled()) {
-						KafkaTemplate.this.logger.trace("Sent ok: " + producerRecord + ", metadata: " + metadata);
-					}
+					KafkaTemplate.this.logger.trace(() -> "Sent ok: " + producerRecord + ", metadata: " + metadata);
 				}
 				else {
 					future.setException(new KafkaProducerException(producerRecord, "Failed to send", exception));
 					if (KafkaTemplate.this.producerListener != null) {
 						KafkaTemplate.this.producerListener.onError(producerRecord, exception);
 					}
-					if (KafkaTemplate.this.logger.isDebugEnabled()) {
-						KafkaTemplate.this.logger.debug("Failed to send: " + producerRecord, exception);
-					}
+					KafkaTemplate.this.logger.debug(exception, () -> "Failed to send: " + producerRecord);
 				}
 			}
 			finally {

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

@@ -25,7 +25,6 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
@@ -38,6 +37,7 @@ import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.event.ContainerStoppedEvent;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
@@ -66,7 +66,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
 
 	private static final int DEFAULT_TOPIC_CHECK_TIMEOUT = 30;
 
-	protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR
+	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass())); // NOSONAR
 
 	protected final ConsumerFactory<K, V> consumerFactory; // NOSONAR (final)
 
@@ -330,7 +330,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
 				}
 			}
 			catch (Exception e) {
-				this.logger.error("Failed to check topic existence", e);
+				this.logger.error(e, "Failed to check topic existence");
 			}
 			if (missing != null && missing.size() > 0) {
 				throw new IllegalStateException(
@@ -368,7 +368,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
 					latch.await(this.containerProperties.getShutdownTimeout(), TimeUnit.MILLISECONDS); // NOSONAR
 					publishContainerStoppedEvent();
 				}
-				catch (InterruptedException e) {
+				catch (@SuppressWarnings("unused") InterruptedException e) {
 					Thread.currentThread().interrupt();
 				}
 			}
@@ -406,18 +406,14 @@ public abstract class AbstractMessageListenerContainer<K, V>
 
 			@Override
 			public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
-				Log logger2 = AbstractMessageListenerContainer.this.logger;
-				if (logger2.isInfoEnabled()) {
-					logger2.info(getGroupId() + ": partitions revoked: " + partitions);
-				}
+				AbstractMessageListenerContainer.this.logger.info(() ->
+					getGroupId() + ": partitions revoked: " + partitions);
 			}
 
 			@Override
 			public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
-				Log logger2 = AbstractMessageListenerContainer.this.logger;
-				if (logger2.isInfoEnabled()) {
-					logger2.info(getGroupId() + ": partitions assigned: " + partitions);
-				}
+				AbstractMessageListenerContainer.this.logger.info(() ->
+					getGroupId() + ": partitions assigned: " + partitions);
 			}
 
 		};

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

@@ -18,10 +18,11 @@ package org.springframework.kafka.listener;
 
 import java.util.Iterator;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 
+import org.springframework.core.log.LogAccessor;
+
 /**
  * Simple handler that invokes a {@link LoggingErrorHandler} for each record.
  *
@@ -31,7 +32,8 @@ import org.apache.kafka.clients.consumer.ConsumerRecords;
  */
 public class BatchLoggingErrorHandler implements BatchErrorHandler {
 
-	private static final Log logger = LogFactory.getLog(BatchLoggingErrorHandler.class); // NOSONAR
+	private static final LogAccessor LOGGER =
+			new LogAccessor(LogFactory.getLog(BatchLoggingErrorHandler.class));
 
 	@Override
 	public void handle(Exception thrownException, ConsumerRecords<?, ?> data) {
@@ -45,7 +47,7 @@ public class BatchLoggingErrorHandler implements BatchErrorHandler {
 				message.append(iterator.next()).append('\n');
 			}
 		}
-		logger.error(message.substring(0, message.length() - 1), thrownException);
+		LOGGER.error(thrownException, () -> message.substring(0, message.length() - 1));
 	}
 
 }

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

@@ -137,7 +137,7 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
 			ContainerProperties containerProperties = getContainerProperties();
 			TopicPartitionInitialOffset[] topicPartitions = containerProperties.getTopicPartitions();
 			if (topicPartitions != null && this.concurrency > topicPartitions.length) {
-				this.logger.warn("When specific partitions are provided, the concurrency must be less than or "
+				this.logger.warn(() -> "When specific partitions are provided, the concurrency must be less than or "
 						+ "equal to the number of partitions; reduced from " + this.concurrency + " to "
 						+ topicPartitions.length);
 				this.concurrency = topicPartitions.length;

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

@@ -26,7 +26,6 @@ import java.util.Optional;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -34,6 +33,7 @@ import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.header.internals.RecordHeader;
 import org.apache.kafka.common.header.internals.RecordHeaders;
 
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.core.KafkaOperations;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.KafkaHeaders;
@@ -52,7 +52,8 @@ import org.springframework.util.ObjectUtils;
  */
 public class DeadLetterPublishingRecoverer implements BiConsumer<ConsumerRecord<?, ?>, Exception> {
 
-	private static final Log logger = LogFactory.getLog(DeadLetterPublishingRecoverer.class); // NOSONAR
+	private static final LogAccessor LOGGER =
+			new LogAccessor(LogFactory.getLog(DeadLetterPublishingRecoverer.class));
 
 	private static final BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition>
 		DEFAULT_DESTINATION_RESOLVER = (cr, e) -> new TopicPartition(cr.topic() + ".DLT", cr.partition());
@@ -140,10 +141,10 @@ public class DeadLetterPublishingRecoverer implements BiConsumer<ConsumerRecord<
 		RecordHeaders headers = new RecordHeaders(record.headers().toArray());
 		enhanceHeaders(headers, record, exception);
 		DeserializationException deserEx = ListenerUtils.getExceptionFromHeader(record,
-				ErrorHandlingDeserializer2.VALUE_DESERIALIZER_EXCEPTION_HEADER, logger);
+				ErrorHandlingDeserializer2.VALUE_DESERIALIZER_EXCEPTION_HEADER, LOGGER);
 		if (deserEx == null) {
 			deserEx = ListenerUtils.getExceptionFromHeader(record,
-					ErrorHandlingDeserializer2.KEY_DESERIALIZER_EXCEPTION_HEADER, logger);
+					ErrorHandlingDeserializer2.KEY_DESERIALIZER_EXCEPTION_HEADER, LOGGER);
 		}
 		ProducerRecord<Object, Object> outRecord = createProducerRecord(record, tp, headers,
 				deserEx == null ? null : deserEx.getData());
@@ -171,9 +172,7 @@ public class DeadLetterPublishingRecoverer implements BiConsumer<ConsumerRecord<
 		if (key.isPresent()) {
 			return (KafkaTemplate<Object, Object>) this.templates.get(key.get());
 		}
-		if (logger.isWarnEnabled()) {
-			logger.warn("Failed to find a template for " + value.getClass() + " attemting to use the last entry");
-		}
+		LOGGER.warn(() -> "Failed to find a template for " + value.getClass() + " attemting to use the last entry");
 		return (KafkaTemplate<Object, Object>) this.templates.values()
 				.stream()
 				.reduce((first,  second) -> second)
@@ -211,15 +210,13 @@ public class DeadLetterPublishingRecoverer implements BiConsumer<ConsumerRecord<
 	protected void publish(ProducerRecord<Object, Object> outRecord, KafkaOperations<Object, Object> kafkaTemplate) {
 		try {
 			kafkaTemplate.send(outRecord).addCallback(result -> {
-				if (logger.isDebugEnabled()) {
-					logger.debug("Successful dead-letter publication: " + result);
-				}
+				LOGGER.debug(() -> "Successful dead-letter publication: " + result);
 			}, ex -> {
-				logger.error("Dead-letter publication failed for: " + outRecord, ex);
+				LOGGER.error(ex, () -> "Dead-letter publication failed for: " + outRecord);
 			});
 		}
 		catch (Exception e) {
-			logger.error("Dead-letter publication failed for: " + outRecord, e);
+			LOGGER.error(e, () -> "Dead-letter publication failed for: " + outRecord);
 		}
 	}
 

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

@@ -20,13 +20,13 @@ import java.util.Collections;
 import java.util.List;
 import java.util.function.BiConsumer;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SeekUtils;
 import org.springframework.lang.Nullable;
@@ -49,7 +49,8 @@ import org.springframework.lang.Nullable;
  */
 public class DefaultAfterRollbackProcessor<K, V> implements AfterRollbackProcessor<K, V> {
 
-	private static final Log logger = LogFactory.getLog(DefaultAfterRollbackProcessor.class); // NOSONAR
+	private static final LogAccessor LOGGER =
+			new LogAccessor(LogFactory.getLog(DefaultAfterRollbackProcessor.class));
 
 	private final FailedRecordTracker failureTracker;
 
@@ -97,7 +98,7 @@ public class DefaultAfterRollbackProcessor<K, V> implements AfterRollbackProcess
 	 */
 	public DefaultAfterRollbackProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer,
 			int maxFailures) {
-		this.failureTracker = new FailedRecordTracker(recoverer, maxFailures, logger);
+		this.failureTracker = new FailedRecordTracker(recoverer, maxFailures, LOGGER);
 	}
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
@@ -105,7 +106,7 @@ public class DefaultAfterRollbackProcessor<K, V> implements AfterRollbackProcess
 	public void process(List<ConsumerRecord<K, V>> records, Consumer<K, V> consumer, Exception exception,
 			boolean recoverable) {
 
-		if (SeekUtils.doSeeks(((List) records), consumer, exception, recoverable, this.failureTracker::skip, logger)
+		if (SeekUtils.doSeeks(((List) records), consumer, exception, recoverable, this.failureTracker::skip, LOGGER)
 				&& this.kafkaTemplate != null && this.kafkaTemplate.isTransactional()) {
 			ConsumerRecord<K, V> skipped = records.get(0);
 			this.kafkaTemplate.sendOffsetsToTransaction(

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

@@ -19,9 +19,9 @@ package org.springframework.kafka.listener;
 import java.time.temporal.ValueRange;
 import java.util.function.BiConsumer;
 
-import org.apache.commons.logging.Log;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
+import org.springframework.core.log.LogAccessor;
 import org.springframework.lang.Nullable;
 
 /**
@@ -41,9 +41,11 @@ class FailedRecordTracker {
 
 	private final boolean noRetries;
 
-	FailedRecordTracker(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, int maxFailures, Log logger) {
+	FailedRecordTracker(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, int maxFailures,
+			LogAccessor logger) {
+
 		if (recoverer == null) {
-			this.recoverer = (r, t) -> logger.error("Max failures (" + maxFailures + ") reached for: " + r, t);
+			this.recoverer = (r, t) -> logger.error(t, "Max failures (" + maxFailures + ") reached for: " + r);
 		}
 		else {
 			this.recoverer = recoverer;

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

@@ -35,7 +35,6 @@ import java.util.concurrent.ScheduledFuture;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -51,6 +50,7 @@ import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.WakeupException;
 
+import org.springframework.core.log.LogAccessor;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.core.ConsumerFactory;
@@ -353,7 +353,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 			}
 		}
 		catch (Exception e) {
-			this.logger.error("Failed to publish consumer stopping event", e);
+			this.logger.error(e, "Failed to publish consumer stopping event");
 		}
 	}
 
@@ -388,7 +388,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 
 		private static final String RAW_TYPES = RAWTYPES;
 
-		private final Log logger = LogFactory.getLog(ListenerConsumer.class);
+		private final LogAccessor logger = new LogAccessor(LogFactory.getLog(ListenerConsumer.class)); // NOSONAR hide
 
 		private final ContainerProperties containerProperties = getContainerProperties();
 
@@ -548,7 +548,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 			this.monitorTask = this.taskScheduler.scheduleAtFixedRate(this::checkConsumer,
 					Duration.ofSeconds(this.containerProperties.getMonitorInterval()));
 			if (this.containerProperties.isLogContainerConfig()) {
-				this.logger.info(this);
+				this.logger.info(this.toString());
 			}
 			Map<String, Object> props = KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties();
 			this.checkNullKeyForExceptions = checkDeserializer(findDeserializerClass(props, false));
@@ -613,8 +613,10 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 					return Duration.ofMillis(Long.parseLong((String) timeout));
 				}
 				else {
-					if (timeout != null && this.logger.isWarnEnabled()) {
-						this.logger.warn("Unexpected type: " + timeout.getClass().getName() + " in property '"
+					if (timeout != null) {
+						Object timeoutToLog = timeout;
+						this.logger.warn(() -> "Unexpected type: " + timeoutToLog.getClass().getName()
+							+ " in property '"
 							+ ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG
 							+ "'; defaulting to 60 seconds for sync commit timeouts");
 					}
@@ -756,7 +758,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 				}
 				catch (NoOffsetForPartitionException nofpe) {
 					this.fatalError = true;
-					ListenerConsumer.this.logger.error("No offset and no reset policy", nofpe);
+					ListenerConsumer.this.logger.error(nofpe, "No offset and no reset policy");
 					break;
 				}
 				catch (Exception e) {
@@ -767,7 +769,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 					if (runnable != null) {
 						runnable.run();
 					}
-					this.logger.error("Stopping container due to an Error", e);
+					this.logger.error(e, "Stopping container due to an Error");
 					wrapUp();
 					throw e;
 				}
@@ -781,7 +783,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 					initPartitionsIfNeeded();
 				}
 				catch (Exception e) {
-					this.logger.error("Failed to set initial offsets", e);
+					this.logger.error(e, "Failed to set initial offsets");
 				}
 			}
 		}
@@ -808,10 +810,10 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 		}
 
 		private void debugRecords(ConsumerRecords<K, V> records) {
-			if (records != null && this.logger.isDebugEnabled()) {
-				this.logger.debug("Received: " + records.count() + " records");
-				if (records.count() > 0 && this.logger.isTraceEnabled()) {
-					this.logger.trace(records.partitions().stream()
+			if (records != null) {
+				this.logger.debug(() -> "Received: " + records.count() + " records");
+				if (records.count() > 0) {
+					this.logger.trace(() -> records.partitions().stream()
 							.flatMap(p -> records.records(p).stream())
 							// map to same format as send metadata toString()
 							.map(r -> r.topic() + "-" + r.partition() + "@" + r.offset())
@@ -824,18 +826,14 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 			if (!this.consumerPaused && isPaused()) {
 				this.consumer.pause(this.consumer.assignment());
 				this.consumerPaused = true;
-				if (this.logger.isDebugEnabled()) {
-					this.logger.debug("Paused consumption from: " + this.consumer.paused());
-				}
+				this.logger.debug(() -> "Paused consumption from: " + this.consumer.paused());
 				publishConsumerPausedEvent(this.consumer.assignment());
 			}
 		}
 
 		private void checkResumed() {
 			if (this.consumerPaused && !isPaused()) {
-				if (this.logger.isDebugEnabled()) {
-					this.logger.debug("Resuming consumption from: " + this.consumer.paused());
-				}
+				this.logger.debug(() -> "Resuming consumption from: " + this.consumer.paused());
 				Set<TopicPartition> paused = this.consumer.paused();
 				this.consumer.resume(paused);
 				this.consumerPaused = false;
@@ -891,9 +889,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 			if (this.errorHandler != null) {
 				this.errorHandler.clearThreadState();
 			}
-			if (this.logger.isInfoEnabled()) {
-				this.logger.info(getGroupId() + ": Consumer stopped");
-			}
+			this.logger.info(() -> getGroupId() + ": Consumer stopped");
 			publishConsumerStoppedEvent();
 		}
 
@@ -913,11 +909,11 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 							KafkaMessageListenerContainer.this);
 				}
 				else {
-					this.logger.error("Consumer exception", e);
+					this.logger.error(e, "Consumer exception");
 				}
 			}
 			catch (Exception ex) {
-				this.logger.error("Consumer exception", ex);
+				this.logger.error(ex, "Consumer exception");
 			}
 		}
 
@@ -935,14 +931,16 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 		private void handleAcks() {
 			ConsumerRecord<K, V> record = this.acks.poll();
 			while (record != null) {
-				if (this.logger.isTraceEnabled()) {
-					this.logger.trace("Ack: " + record);
-				}
+				traceAck(record);
 				processAck(record);
 				record = this.acks.poll();
 			}
 		}
 
+		private void traceAck(ConsumerRecord<K, V> record) {
+			this.logger.trace(() -> "Ack: " + record);
+		}
+
 		private void processAck(ConsumerRecord<K, V> record) {
 			if (!Thread.currentThread().equals(this.consumerThread)) {
 				try {
@@ -1028,7 +1026,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 				});
 			}
 			catch (RuntimeException e) {
-				this.logger.error("Transaction rolled back", e);
+				this.logger.error(e, "Transaction rolled back");
 				AfterRollbackProcessor<K, V> afterRollbackProcessorToUse =
 						(AfterRollbackProcessor<K, V>) getAfterRollbackProcessor();
 				if (afterRollbackProcessorToUse.isProcessInTransaction() && this.transactionTemplate != null) {
@@ -1090,11 +1088,11 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 					invokeBatchErrorHandler(records, producer, e);
 				}
 				catch (RuntimeException ee) {
-					this.logger.error("Error handler threw an exception", ee);
+					this.logger.error(ee, "Error handler threw an exception");
 					return ee;
 				}
 				catch (Error er) { // NOSONAR
-					this.logger.error("Error handler threw an error", er);
+					this.logger.error(er, "Error handler threw an error");
 					throw er;
 				}
 			}
@@ -1185,9 +1183,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 			Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
 			while (iterator.hasNext()) {
 				final ConsumerRecord<K, V> record = iterator.next();
-				if (this.logger.isTraceEnabled()) {
-					this.logger.trace("Processing " + record);
-				}
+				this.logger.trace(() -> "Processing " + record);
 				try {
 					TransactionSupport
 							.setTransactionIdSuffix(zombieFenceTxIdSuffix(record.topic(), record.partition()));
@@ -1210,7 +1206,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 					});
 				}
 				catch (RuntimeException e) {
-					this.logger.error("Transaction rolled back", e);
+					this.logger.error(e, "Transaction rolled back");
 					recordAfterRollback(iterator, record, e);
 				}
 				finally {
@@ -1249,9 +1245,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 			Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
 			while (iterator.hasNext()) {
 				final ConsumerRecord<K, V> record = iterator.next();
-				if (this.logger.isTraceEnabled()) {
-					this.logger.trace("Processing " + record);
-				}
+				this.logger.trace(() -> "Processing " + record);
 				doInvokeRecordListener(record, null, iterator);
 			}
 		}
@@ -1284,11 +1278,11 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 					invokeErrorHandler(record, producer, iterator, e);
 				}
 				catch (RuntimeException ee) {
-					this.logger.error("Error handler threw an exception", ee);
+					this.logger.error(ee, "Error handler threw an exception");
 					return ee;
 				}
 				catch (Error er) { // NOSONAR
-					this.logger.error("Error handler threw an error", er);
+					this.logger.error(er, "Error handler threw an error");
 					throw er;
 				}
 			}
@@ -1412,7 +1406,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 					sendOffsetsToTransaction(producer);
 				}
 				catch (Exception e) {
-					this.logger.error("Send offsets to transaction failed", e);
+					this.logger.error(e, "Send offsets to transaction failed");
 				}
 			}
 		}
@@ -1435,8 +1429,8 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 				}
 				boolean countExceeded = this.isCountAck && this.count >= this.containerProperties.getAckCount();
 				if ((!this.isTimeOnlyAck && !this.isCountAck) || countExceeded) {
-					if (this.logger.isDebugEnabled() && isCountAck) {
-						this.logger.debug("Committing in " + ackMode.name() + " because count "
+					if (this.isCountAck) {
+						this.logger.debug(() -> "Committing in " + ackMode.name() + " because count "
 								+ this.count
 								+ " exceeds configured limit of " + this.containerProperties.getAckCount());
 					}
@@ -1454,20 +1448,16 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 			now = System.currentTimeMillis();
 			boolean elapsed = now - this.last > this.containerProperties.getAckTime();
 			if (ackMode.equals(AckMode.TIME) && elapsed) {
-				if (this.logger.isDebugEnabled()) {
-					this.logger.debug("Committing in AckMode.TIME " +
-							"because time elapsed exceeds configured limit of " +
-							this.containerProperties.getAckTime());
-				}
+				this.logger.debug(() -> "Committing in AckMode.TIME " +
+						"because time elapsed exceeds configured limit of " +
+						this.containerProperties.getAckTime());
 				commitIfNecessary();
 				this.last = now;
 			}
 			else if (ackMode.equals(AckMode.COUNT_TIME) && elapsed) {
-				if (this.logger.isDebugEnabled()) {
-					this.logger.debug("Committing in AckMode.COUNT_TIME " +
-							"because time elapsed exceeds configured limit of " +
-							this.containerProperties.getAckTime());
-				}
+				this.logger.debug(() -> "Committing in AckMode.COUNT_TIME " +
+						"because time elapsed exceeds configured limit of " +
+						this.containerProperties.getAckTime());
 				commitIfNecessary();
 				this.last = now;
 				this.count = 0;
@@ -1477,9 +1467,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 		private void processSeeks() {
 			TopicPartitionInitialOffset offset = this.seeks.poll();
 			while (offset != null) {
-				if (this.logger.isTraceEnabled()) {
-					this.logger.trace("Seek: " + offset);
-				}
+				traceSeek(offset);
 				try {
 					SeekPosition position = offset.getPosition();
 					if (position == null) {
@@ -1493,12 +1481,17 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 					}
 				}
 				catch (Exception e) {
-					this.logger.error("Exception while seeking " + offset, e);
+					TopicPartitionInitialOffset offsetToLog = offset;
+					this.logger.error(e, () -> "Exception while seeking " + offsetToLog);
 				}
 				offset = this.seeks.poll();
 			}
 		}
 
+		private void traceSeek(TopicPartitionInitialOffset offset) {
+			this.logger.trace(() -> "Seek: " + offset);
+		}
+
 		private void initPartitionsIfNeeded() {
 			/*
 			 * Note: initial position setting is only supported with explicit topic assignment.
@@ -1542,18 +1535,21 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 
 					try {
 						this.consumer.seek(topicPartition, newOffset);
-						if (this.logger.isDebugEnabled()) {
-							this.logger.debug("Reset " + topicPartition + " to offset " + newOffset);
-						}
+						logReset(topicPartition, newOffset);
 					}
 					catch (Exception e) {
-						this.logger.error("Failed to set initial offset for " + topicPartition
-								+ " at " + newOffset + ". Position is " + this.consumer.position(topicPartition), e);
+						long newOffsetToLog = newOffset;
+						this.logger.error(e, () -> "Failed to set initial offset for " + topicPartition
+								+ " at " + newOffsetToLog + ". Position is " + this.consumer.position(topicPartition));
 					}
 				}
 			}
 		}
 
+		private void logReset(TopicPartition topicPartition, long newOffset) {
+			this.logger.debug(() -> "Reset " + topicPartition + " to offset " + newOffset);
+		}
+
 		private void updatePendingOffsets() {
 			ConsumerRecord<K, V> record = this.acks.poll();
 			while (record != null) {
@@ -1569,9 +1565,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 
 		private void commitIfNecessary() {
 			Map<TopicPartition, OffsetAndMetadata> commits = buildCommits();
-			if (this.logger.isDebugEnabled()) {
-				this.logger.debug("Commit list: " + commits);
-			}
+			this.logger.debug(() -> "Commit list: " + commits);
 			if (!commits.isEmpty()) {
 				this.commitLogger.log(() -> "Committing: " + commits);
 				try {
@@ -1647,8 +1641,8 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 						producerFactory.closeProducerFor(zombieFenceTxIdSuffix(tp.topic(), tp.partition()));
 					}
 					catch (Exception e) {
-						this.logger.error("Failed to close producer with transaction id suffix: "
-								+ zombieFenceTxIdSuffix(tp.topic(), tp.partition()), e);
+						this.logger.error(e, () -> "Failed to close producer with transaction id suffix: "
+								+ zombieFenceTxIdSuffix(tp.topic(), tp.partition()));
 					}
 				});
 			}
@@ -1764,7 +1758,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 						}
 						catch (NoOffsetForPartitionException e) {
 							ListenerConsumer.this.fatalError = true;
-							ListenerConsumer.this.logger.error("No offset and no reset policy", e);
+							ListenerConsumer.this.logger.error(e, "No offset and no reset policy");
 							return;
 						}
 					}
@@ -1855,7 +1849,8 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 
 		@Override
 		public void onFailure(Throwable e) {
-			KafkaMessageListenerContainer.this.logger.error("Error while stopping the container: ", e);
+			KafkaMessageListenerContainer.this.logger
+					.error(e, "Error while stopping the container: ");
 			if (this.callback != null) {
 				this.callback.run();
 			}
@@ -1863,10 +1858,8 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 
 		@Override
 		public void onSuccess(Object result) {
-			if (KafkaMessageListenerContainer.this.logger.isDebugEnabled()) {
-				KafkaMessageListenerContainer.this.logger
-						.debug(KafkaMessageListenerContainer.this + " stopped normally");
-			}
+			KafkaMessageListenerContainer.this.logger
+					.debug(() -> KafkaMessageListenerContainer.this + " stopped normally");
 			if (this.callback != null) {
 				this.callback.run();
 			}

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

@@ -22,12 +22,12 @@ import java.io.ObjectInputStream;
 import java.util.Arrays;
 import java.util.stream.Collectors;
 
-import org.apache.commons.logging.Log;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeaders;
 
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.support.serializer.DeserializationException;
 import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2;
 import org.springframework.lang.Nullable;
@@ -81,7 +81,7 @@ public final class ListenerUtils {
 	 */
 	@Nullable
 	public static DeserializationException getExceptionFromHeader(final ConsumerRecord<?, ?> record,
-			String headerName, Log logger) {
+			String headerName, LogAccessor logger) {
 
 		Header header = record.headers().lastHeader(headerName);
 		if (header != null) {
@@ -96,7 +96,7 @@ public final class ListenerUtils {
 				return ex;
 			}
 			catch (IOException | ClassNotFoundException | ClassCastException e) {
-				logger.error("Failed to deserialize a deserialization exception", e);
+				logger.error(e, "Failed to deserialize a deserialization exception");
 			}
 		}
 		return null;

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

@@ -18,12 +18,13 @@ package org.springframework.kafka.listener;
 
 import java.util.Map;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.consumer.OffsetCommitCallback;
 import org.apache.kafka.common.TopicPartition;
 
+import org.springframework.core.log.LogAccessor;
+
 /**
  * Logs commit results at DEBUG level for success and ERROR for failures.
  *
@@ -32,15 +33,15 @@ import org.apache.kafka.common.TopicPartition;
  */
 public final class LoggingCommitCallback implements OffsetCommitCallback {
 
-	private static final Log logger = LogFactory.getLog(LoggingCommitCallback.class); // NOSONAR
+	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(LoggingCommitCallback.class));
 
 	@Override
 	public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
 		if (exception != null) {
-			logger.error("Commit failed for " + offsets, exception);
+			LOGGER.error(exception, () -> "Commit failed for " + offsets);
 		}
-		else if (logger.isDebugEnabled()) {
-			logger.debug("Commits for " + offsets + " completed");
+		else {
+			LOGGER.debug(() -> "Commits for " + offsets + " completed");
 		}
 	}
 

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

@@ -16,10 +16,10 @@
 
 package org.springframework.kafka.listener;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
+import org.springframework.core.log.LogAccessor;
 import org.springframework.util.ObjectUtils;
 
 /**
@@ -30,11 +30,11 @@ import org.springframework.util.ObjectUtils;
  */
 public class LoggingErrorHandler implements ErrorHandler {
 
-	private static final Log logger = LogFactory.getLog(LoggingErrorHandler.class); // NOSONAR
+	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(LoggingErrorHandler.class));
 
 	@Override
 	public void handle(Exception thrownException, ConsumerRecord<?, ?> record) {
-		logger.error("Error while processing: " + ObjectUtils.nullSafeToString(record), thrownException);
+		LOGGER.error(thrownException, () -> "Error while processing: " + ObjectUtils.nullSafeToString(record));
 	}
 
 }

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

@@ -23,7 +23,6 @@ import java.util.Map;
 import java.util.function.BiConsumer;
 import java.util.function.BiPredicate;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -32,6 +31,7 @@ import org.apache.kafka.clients.consumer.OffsetCommitCallback;
 import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.classify.BinaryExceptionClassifier;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.listener.ContainerProperties.AckMode;
 import org.springframework.kafka.support.SeekUtils;
@@ -56,7 +56,8 @@ public class SeekToCurrentErrorHandler implements ContainerAwareErrorHandler {
 
 	private static final BiPredicate<ConsumerRecord<?, ?>, Exception> ALWAYS_SKIP_PREDICATE = (r, e) -> true;
 
-	protected static final Log LOGGER = LogFactory.getLog(SeekToCurrentErrorHandler.class); // NOSONAR visibility
+	protected static final LogAccessor LOGGER =
+			new LogAccessor(LogFactory.getLog(SeekToCurrentErrorHandler.class)); // NOSONAR visibility
 
 	private static final LoggingCommitCallback LOGGING_COMMIT_CALLBACK = new LoggingCommitCallback();
 
@@ -236,7 +237,8 @@ public class SeekToCurrentErrorHandler implements ContainerAwareErrorHandler {
 				}
 			}
 			else {
-				LOGGER.warn("'commitRecovered' ignored, container AckMode must be MANUAL_IMMEDIATE");
+				LOGGER.warn(() -> "'commitRecovered' ignored, container AckMode must be MANUAL_IMMEDIATE, not "
+						+ container.getContainerProperties().getAckMode());
 			}
 		}
 	}

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractDelegatingMessageListenerAdapter.java

@@ -18,10 +18,10 @@ package org.springframework.kafka.listener.adapter;
 
 import java.util.Map;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.common.TopicPartition;
 
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.listener.ConsumerSeekAware;
 import org.springframework.kafka.listener.DelegatingMessageListener;
 import org.springframework.kafka.listener.ListenerType;
@@ -39,7 +39,7 @@ import org.springframework.kafka.listener.ListenerUtils;
 public abstract class AbstractDelegatingMessageListenerAdapter<T>
 		implements ConsumerSeekAware, DelegatingMessageListener<T> {
 
-	protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR
+	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass())); // NOSONAR
 
 	protected final T delegate; //NOSONAR
 

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java

@@ -132,9 +132,7 @@ public class BatchMessagingMessageListenerAdapter<K, V> extends MessagingMessage
 		else {
 			message = NULL_MESSAGE; // optimization since we won't need any conversion to invoke
 		}
-		if (logger.isDebugEnabled()) {
-			logger.debug("Processing [" + message + "]");
-		}
+		logger.debug(() -> "Processing [" + message + "]");
 		invoke(records, acknowledgment, consumer, message);
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java

@@ -27,7 +27,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -36,6 +35,7 @@ import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.context.expression.MapAccessor;
 import org.springframework.core.MethodParameter;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.expression.BeanResolver;
 import org.springframework.expression.Expression;
 import org.springframework.expression.ParserContext;
@@ -85,7 +85,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
 
 	private final Object bean;
 
-	protected final Log logger = LogFactory.getLog(getClass()); //NOSONAR
+	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); //NOSONAR
 
 	private final Type inferredType;
 
@@ -314,10 +314,8 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
 	 * {@code o.s.messaging.Message<?>}; may be null
 	 */
 	protected void handleResult(Object resultArg, Object request, Object source) {
-		if (this.logger.isDebugEnabled()) {
-			this.logger.debug("Listener method returned result [" + resultArg
-					+ "] - generating response message for it");
-		}
+		this.logger.debug(() -> "Listener method returned result [" + resultArg
+				+ "] - generating response message for it");
 		boolean isInvocationResult = resultArg instanceof InvocationResult;
 		Object result = isInvocationResult ? ((InvocationResult) resultArg).getResult() : resultArg;
 		String replyTopic = evaluateReplyTopic(request, source, resultArg);
@@ -376,15 +374,13 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
 	 * @param result the result.
 	 * @param topic the topic.
 	 * @param source the source (input).
-	 * @param messageReturnType true if we are returning message(s).
+	 * @param returnTypeMessage true if we are returning message(s).
 	 * @since 2.1.3
 	 */
 	@SuppressWarnings("unchecked")
-	protected void sendResponse(Object result, String topic, @Nullable Object source, boolean messageReturnType) {
-		if (!messageReturnType && topic == null) {
-			if (this.logger.isDebugEnabled()) {
-				this.logger.debug("No replyTopic to handle the reply: " + result);
-			}
+	protected void sendResponse(Object result, String topic, @Nullable Object source, boolean returnTypeMessage) {
+		if (!returnTypeMessage && topic == null) {
+			this.logger.debug(() -> "No replyTopic to handle the reply: " + result);
 		}
 		else if (result instanceof Message) {
 			this.replyTemplate.send((Message<?>) result);
@@ -492,10 +488,8 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
 					genericParameterType = extractGenericParameterTypFromMethodParameter(methodParameter);
 				}
 				else {
-					if (this.logger.isDebugEnabled()) {
-						this.logger.debug("Ambiguous parameters for target payload for method " + method
-								+ "; no inferred type available");
-					}
+					this.logger.debug(() -> "Ambiguous parameters for target payload for method " + method
+							+ "; no inferred type available");
 					break;
 				}
 			}

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordMessagingMessageListenerAdapter.java

@@ -72,9 +72,7 @@ public class RecordMessagingMessageListenerAdapter<K, V> extends MessagingMessag
 	@Override
 	public void onMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
 		Message<?> message = toMessagingMessage(record, acknowledgment, consumer);
-		if (logger.isDebugEnabled()) {
-			logger.debug("Processing [" + message + "]");
-		}
+		logger.debug(() -> "Processing [" + message + "]");
 		try {
 			Object result = invokeHandler(record, acknowledgment, message, consumer);
 			if (result != null) {

spring-kafka/src/main/java/org/springframework/kafka/requestreply/AggregatingReplyingKafkaTemplate.java

@@ -122,7 +122,7 @@ public class AggregatingReplyingKafkaTemplate<K, V, R>
 		data.forEach(record -> {
 			Header correlation = record.headers().lastHeader(KafkaHeaders.CORRELATION_ID);
 			if (correlation == null) {
-				this.logger.error("No correlationId found in reply: " + record
+				this.logger.error(() -> "No correlationId found in reply: " + record
 						+ " - to use request/reply semantics, the responding server must return the correlation id "
 						+ " in the '" + KafkaHeaders.CORRELATION_ID + "' header");
 			}

spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java

@@ -121,7 +121,7 @@ public class ReplyingKafkaTemplate<K, V, R> extends KafkaTemplate<K, V> implemen
 		if (tempReplyTopic == null) {
 			this.replyTopic = null;
 			this.replyPartition = null;
-			this.logger.debug("Could not determine container's reply topic/partition; senders must populate "
+			this.logger.debug(() -> "Could not determine container's reply topic/partition; senders must populate "
 					+ "at least the " + KafkaHeaders.REPLY_TOPIC + " header, and optionally the "
 					+ KafkaHeaders.REPLY_PARTITION + " header");
 		}
@@ -255,9 +255,7 @@ public class ReplyingKafkaTemplate<K, V, R> extends KafkaTemplate<K, V> implemen
 			}
 		}
 		headers.add(new RecordHeader(KafkaHeaders.CORRELATION_ID, correlationId.getCorrelationId()));
-		if (this.logger.isDebugEnabled()) {
-			this.logger.debug("Sending: " + record + WITH_CORRELATION_ID + correlationId);
-		}
+		this.logger.debug(() -> "Sending: " + record + WITH_CORRELATION_ID + correlationId);
 		RequestReplyFuture<K, V, R> future = new RequestReplyFuture<>();
 		this.futures.put(correlationId, future);
 		try {
@@ -275,9 +273,7 @@ public class ReplyingKafkaTemplate<K, V, R> extends KafkaTemplate<K, V> implemen
 		this.scheduler.schedule(() -> {
 			RequestReplyFuture<K, V, R> removed = this.futures.remove(correlationId);
 			if (removed != null) {
-				if (this.logger.isWarnEnabled()) {
-					this.logger.warn("Reply timed out for: " + record + WITH_CORRELATION_ID + correlationId);
-				}
+				this.logger.warn(() -> "Reply timed out for: " + record + WITH_CORRELATION_ID + correlationId);
 				if (!handleTimeout(correlationId, removed)) {
 					removed.setException(new KafkaReplyTimeoutException("Reply timed out"));
 				}
@@ -351,19 +347,18 @@ public class ReplyingKafkaTemplate<K, V, R> extends KafkaTemplate<K, V> implemen
 				}
 			}
 			if (correlationId == null) {
-				this.logger.error("No correlationId found in reply: " + record
+				this.logger.error(() -> "No correlationId found in reply: " + record
 						+ " - to use request/reply semantics, the responding server must return the correlation id "
 						+ " in the '" + KafkaHeaders.CORRELATION_ID + "' header");
 			}
 			else {
 				RequestReplyFuture<K, V, R> future = this.futures.remove(correlationId);
+				CorrelationKey correlationKey = correlationId;
 				if (future == null) {
 					logLateArrival(record, correlationId);
 				}
 				else {
-					if (this.logger.isDebugEnabled()) {
-						this.logger.debug("Received: " + record + WITH_CORRELATION_ID + correlationId);
-					}
+					this.logger.debug(() -> "Received: " + record + WITH_CORRELATION_ID + correlationKey);
 					future.set(record);
 				}
 			}
@@ -372,12 +367,10 @@ public class ReplyingKafkaTemplate<K, V, R> extends KafkaTemplate<K, V> implemen
 
 	protected void logLateArrival(ConsumerRecord<K, R> record, CorrelationKey correlationId) {
 		if (this.sharedReplyTopic) {
-			if (this.logger.isDebugEnabled()) {
-				this.logger.debug(missingCorrelationLogMessage(record, correlationId));
-			}
+			this.logger.debug(() -> missingCorrelationLogMessage(record, correlationId));
 		}
-		else if (this.logger.isErrorEnabled()) {
-			this.logger.error(missingCorrelationLogMessage(record, correlationId));
+		else {
+			this.logger.error(() -> missingCorrelationLogMessage(record, correlationId));
 		}
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java

@@ -28,10 +28,10 @@ import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.common.header.Header;
 
+import org.springframework.core.log.LogAccessor;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.util.Assert;
@@ -49,7 +49,7 @@ import org.springframework.util.PatternMatchUtils;
  */
 public abstract class AbstractKafkaHeaderMapper implements KafkaHeaderMapper {
 
-	protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR
+	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); // NOSONAR
 
 	private final List<HeaderMatcher> matchers = new ArrayList<>();
 
@@ -141,10 +141,8 @@ public abstract class AbstractKafkaHeaderMapper implements KafkaHeaderMapper {
 		if (matches(header)) {
 			if ((header.equals(MessageHeaders.REPLY_CHANNEL) || header.equals(MessageHeaders.ERROR_CHANNEL))
 					&& !(value instanceof String)) {
-				if (this.logger.isDebugEnabled()) {
-					this.logger.debug("Cannot map " + header + " when type is [" + value.getClass()
-							+ "]; it must be a String");
-				}
+				this.logger.debug(() -> "Cannot map " + header + " when type is [" + value.getClass()
+						+ "]; it must be a String");
 				return false;
 			}
 			return true;
@@ -158,10 +156,8 @@ public abstract class AbstractKafkaHeaderMapper implements KafkaHeaderMapper {
 				return !matcher.isNegated();
 			}
 		}
-		if (this.logger.isDebugEnabled()) {
-			this.logger.debug(MessageFormat.format("headerName=[{0}] WILL NOT be mapped; matched no patterns",
-					header));
-		}
+		this.logger.debug(() -> MessageFormat.format("headerName=[{0}] WILL NOT be mapped; matched no patterns",
+				header));
 		return false;
 	}
 
@@ -270,7 +266,8 @@ public abstract class AbstractKafkaHeaderMapper implements KafkaHeaderMapper {
 	 */
 	protected static class SimplePatternBasedHeaderMatcher implements HeaderMatcher {
 
-		private static final Log LOGGER = LogFactory.getLog(SimplePatternBasedHeaderMatcher.class);
+		private static final LogAccessor LOGGER =
+				new LogAccessor(LogFactory.getLog(SimplePatternBasedHeaderMatcher.class));
 
 		private final String pattern;
 
@@ -290,13 +287,11 @@ public abstract class AbstractKafkaHeaderMapper implements KafkaHeaderMapper {
 		public boolean matchHeader(String headerName) {
 			String header = headerName.toLowerCase();
 			if (PatternMatchUtils.simpleMatch(this.pattern, header)) {
-				if (LOGGER.isDebugEnabled()) {
-					LOGGER.debug(
-							MessageFormat.format(
-									"headerName=[{0}] WILL " + (this.negate ? "NOT " : "")
-											+ "be mapped, matched pattern=" + (this.negate ? "!" : "") + "{1}",
-									headerName, this.pattern));
-				}
+				LOGGER.debug(() ->
+						MessageFormat.format(
+								"headerName=[{0}] WILL " + (this.negate ? "NOT " : "")
+										+ "be mapped, matched pattern=" + (this.negate ? "!" : "") + "{1}",
+								headerName, this.pattern));
 				return true;
 			}
 			return false;

spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java

@@ -229,9 +229,7 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper {
 						jsonHeaders.put(key, className);
 					}
 					catch (Exception e) {
-						if (logger.isDebugEnabled()) {
-							logger.debug("Could not map " + key + " with type " + valueToAdd.getClass().getName(), e);
-						}
+						logger.debug(e, () -> "Could not map " + key + " with type " + valueToAdd.getClass().getName());
 					}
 				}
 			}
@@ -241,7 +239,7 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper {
 				target.add(new RecordHeader(JSON_TYPES, headerObjectMapper.writeValueAsBytes(jsonHeaders)));
 			}
 			catch (IllegalStateException | JsonProcessingException e) {
-				logger.error("Could not add json types header", e);
+				logger.error(e, "Could not add json types header");
 			}
 		}
 	}
@@ -262,7 +260,7 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper {
 						}
 					}
 					catch (Exception e) {
-						logger.error("Could not load class for header: " + header.key(), e);
+						logger.error(e, () -> "Could not load class for header: " + header.key());
 					}
 					if (trusted) {
 						try {
@@ -270,8 +268,9 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper {
 							headers.put(header.key(), value);
 						}
 						catch (IOException e) {
-							logger.error("Could not decode json type: " + new String(header.value()) + " for key: " +
-									header.key(), e);
+							logger.error(e, () ->
+								"Could not decode json type: " + new String(header.value()) + " for key: "
+								+ header.key());
 							headers.put(header.key(), header.value());
 						}
 					}
@@ -298,7 +297,7 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper {
 							ClassUtils.forName(nth.getUntrustedType(), null));
 				}
 				catch (Exception e) {
-					logger.error("Could not decode header: " + nth, e);
+					logger.error(e, () -> "Could not decode header: " + nth);
 				}
 			}
 		}
@@ -318,7 +317,7 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper {
 					types = headerObjectMapper.readValue(next.value(), Map.class);
 				}
 				catch (IOException e) {
-					logger.error("Could not decode json types: " + new String(next.value()), e);
+					logger.error(e, () -> "Could not decode json types: " + new String(next.value()));
 				}
 				break;
 			}

spring-kafka/src/main/java/org/springframework/kafka/support/LogIfLevelEnabled.java

@@ -18,8 +18,7 @@ package org.springframework.kafka.support;
 
 import java.util.function.Supplier;
 
-import org.apache.commons.logging.Log;
-
+import org.springframework.core.log.LogAccessor;
 import org.springframework.util.Assert;
 
 /**
@@ -32,11 +31,11 @@ import org.springframework.util.Assert;
  */
 public final class LogIfLevelEnabled {
 
-	private final Log logger;
+	private final LogAccessor logger;
 
 	private final Level level;
 
-	public LogIfLevelEnabled(Log logger, Level level) {
+	public LogIfLevelEnabled(LogAccessor logger, Level level) {
 		Assert.notNull(logger, "'logger' cannot be null");
 		Assert.notNull(level, "'level' cannot be null");
 		this.logger = logger;
@@ -80,7 +79,7 @@ public final class LogIfLevelEnabled {
 
 	}
 
-	public void log(Supplier<Object> messageSupplier) {
+	public void log(Supplier<CharSequence> messageSupplier) {
 		switch (this.level) {
 			case FATAL:
 				fatal(messageSupplier, null);
@@ -103,92 +102,80 @@ public final class LogIfLevelEnabled {
 		}
 	}
 
-	public void log(Supplier<Object> messageSupplier, Throwable t) {
+	public void log(Supplier<CharSequence> messageSupplier, Throwable thrown) {
 		switch (this.level) {
 			case FATAL:
-			fatal(messageSupplier, t);
+			fatal(messageSupplier, thrown);
 				break;
 			case ERROR:
-			error(messageSupplier, t);
+			error(messageSupplier, thrown);
 				break;
 			case WARN:
-			warn(messageSupplier, t);
+			warn(messageSupplier, thrown);
 				break;
 			case INFO:
-			info(messageSupplier, t);
+			info(messageSupplier, thrown);
 				break;
 			case DEBUG:
-			debug(messageSupplier, t);
+			debug(messageSupplier, thrown);
 				break;
 			case TRACE:
-			trace(messageSupplier, t);
+			trace(messageSupplier, thrown);
 				break;
 		}
 	}
 
-	private void fatal(Supplier<Object> messageSupplier, Throwable t) {
-		if (this.logger.isFatalEnabled()) {
-			if (t != null) {
-				this.logger.fatal(messageSupplier.get(), t);
-			}
-			else {
-				this.logger.fatal(messageSupplier.get());
-			}
+	private void fatal(Supplier<CharSequence> messageSupplier, Throwable thrown) {
+		if (thrown != null) {
+			this.logger.fatal(thrown, messageSupplier);
+		}
+		else {
+			this.logger.fatal(messageSupplier);
 		}
 	}
 
-	private void error(Supplier<Object> messageSupplier, Throwable t) {
-		if (this.logger.isErrorEnabled()) {
-			if (t != null) {
-				this.logger.error(messageSupplier.get(), t);
-			}
-			else {
-				this.logger.error(messageSupplier.get());
-			}
+	private void error(Supplier<CharSequence> messageSupplier, Throwable thrown) {
+		if (thrown != null) {
+			this.logger.error(thrown, messageSupplier);
+		}
+		else {
+			this.logger.error(messageSupplier);
 		}
 	}
 
-	private void warn(Supplier<Object> messageSupplier, Throwable t) {
-		if (this.logger.isWarnEnabled()) {
-			if (t != null) {
-				this.logger.warn(messageSupplier.get(), t);
-			}
-			else {
-				this.logger.warn(messageSupplier.get());
-			}
+	private void warn(Supplier<CharSequence> messageSupplier, Throwable thrown) {
+		if (thrown != null) {
+			this.logger.warn(thrown, messageSupplier);
+		}
+		else {
+			this.logger.warn(messageSupplier);
 		}
 	}
 
-	private void info(Supplier<Object> messageSupplier, Throwable t) {
-		if (this.logger.isInfoEnabled()) {
-			if (t != null) {
-				this.logger.info(messageSupplier.get(), t);
-			}
-			else {
-				this.logger.info(messageSupplier.get());
-			}
+	private void info(Supplier<CharSequence> messageSupplier, Throwable thrown) {
+		if (thrown != null) {
+			this.logger.info(thrown, messageSupplier);
+		}
+		else {
+			this.logger.info(messageSupplier);
 		}
 	}
 
-	private void debug(Supplier<Object> messageSupplier, Throwable t) {
-		if (this.logger.isDebugEnabled()) {
-			if (t != null) {
-				this.logger.debug(messageSupplier.get(), t);
-			}
-			else {
-				this.logger.debug(messageSupplier.get());
-			}
+	private void debug(Supplier<CharSequence> messageSupplier, Throwable thrown) {
+		if (thrown != null) {
+			this.logger.debug(thrown, messageSupplier);
+		}
+		else {
+			this.logger.debug(messageSupplier);
 		}
 	}
 
-	private void trace(Supplier<Object> messageSupplier, Throwable t) {
-		if (this.logger.isTraceEnabled()) {
-			if (t != null) {
-				this.logger.trace(messageSupplier.get(), t);
-			}
-			else {
-				this.logger.trace(messageSupplier.get());
-			}
+	private void trace(Supplier<CharSequence> messageSupplier, Throwable thrown) {
+		if (thrown != null) {
+			this.logger.trace(thrown, messageSupplier);
+		}
+		else {
+			this.logger.trace(messageSupplier);
 		}
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/support/LoggingProducerListener.java

@@ -16,9 +16,9 @@
 
 package org.springframework.kafka.support;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import org.springframework.core.log.LogAccessor;
 import org.springframework.util.ObjectUtils;
 
 /**
@@ -37,7 +37,7 @@ public class LoggingProducerListener<K, V> implements ProducerListener<K, V> {
 	 */
 	public static final int DEFAULT_MAX_CONTENT_LOGGED = 100;
 
-	private static final Log logger = LogFactory.getLog(LoggingProducerListener.class); // NOSONAR
+	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(LoggingProducerListener.class));
 
 	private boolean includeContents = true;
 
@@ -65,7 +65,7 @@ public class LoggingProducerListener<K, V> implements ProducerListener<K, V> {
 
 	@Override
 	public void onError(String topic, Integer partition, K key, V value, Exception exception) {
-		if (logger.isErrorEnabled()) {
+		LOGGER.error(exception, () -> {
 			StringBuffer logOutput = new StringBuffer();
 			logOutput.append("Exception thrown when sending a message");
 			if (this.includeContents) {
@@ -81,8 +81,8 @@ public class LoggingProducerListener<K, V> implements ProducerListener<K, V> {
 				logOutput.append(" and partition ").append(partition);
 			}
 			logOutput.append(":");
-			logger.error(logOutput, exception);
-		}
+			return logOutput.toString();
+		});
 	}
 
 	private String toDisplayString(String original, int maxCharacters) {

spring-kafka/src/main/java/org/springframework/kafka/support/SeekUtils.java

@@ -22,11 +22,12 @@ import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiPredicate;
 
-import org.apache.commons.logging.Log;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.TopicPartition;
 
+import org.springframework.core.log.LogAccessor;
+
 /**
  * Seek utilities.
  *
@@ -56,7 +57,7 @@ public final class SeekUtils {
 	 * @return true if the failed record was skipped.
 	 */
 	public static boolean doSeeks(List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer, Exception exception,
-			boolean recoverable, BiPredicate<ConsumerRecord<?, ?>, Exception> skipper, Log logger) {
+			boolean recoverable, BiPredicate<ConsumerRecord<?, ?>, Exception> skipper, LogAccessor logger) {
 
 		Map<TopicPartition, Long> partitions = new LinkedHashMap<>();
 		AtomicBoolean first = new AtomicBoolean(true);
@@ -64,8 +65,8 @@ public final class SeekUtils {
 		records.forEach(record ->  {
 			if (recoverable && first.get()) {
 				skipped.set(skipper.test(record, exception));
-				if (skipped.get() && logger.isDebugEnabled()) {
-					logger.debug("Skipping seek of: " + record);
+				if (skipped.get()) {
+					logger.debug(() -> "Skipping seek of: " + record);
 				}
 			}
 			if (!recoverable || !first.get() || !skipped.get()) {
@@ -74,16 +75,13 @@ public final class SeekUtils {
 			}
 			first.set(false);
 		});
-		boolean tracing = logger.isTraceEnabled();
 		partitions.forEach((topicPartition, offset) -> {
 			try {
-				if (tracing) {
-					logger.trace("Seeking: " + topicPartition + " to: " + offset);
-				}
+				logger.trace(() -> "Seeking: " + topicPartition + " to: " + offset);
 				consumer.seek(topicPartition, offset);
 			}
 			catch (Exception e) {
-				logger.error("Failed to seek " + topicPartition + " to " + offset, e);
+				logger.error(e, () -> "Failed to seek " + topicPartition + " to " + offset);
 			}
 		});
 		return skipped.get();

spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java

@@ -23,13 +23,13 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.header.Headers;
 
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
 import org.springframework.kafka.support.JacksonPresent;
@@ -60,7 +60,7 @@ import org.springframework.messaging.support.MessageBuilder;
  */
 public class BatchMessagingMessageConverter implements BatchMessageConverter {
 
-	protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR
+	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); // NOSONAR
 
 	private final RecordMessageConverter recordConverter;
 
@@ -167,8 +167,8 @@ public class BatchMessagingMessageConverter implements BatchMessageConverter {
 				convertedHeaders.add(converted);
 			}
 			else {
-				if (this.logger.isDebugEnabled() && !logged) {
-					this.logger.debug(
+				if (!logged) {
+					this.logger.debug(() ->
 						"No header mapper is available; Jackson is required for the default mapper; "
 						+ "headers (if present) are not mapped but provided raw in "
 						+ KafkaHeaders.NATIVE_HEADERS);

spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java

@@ -20,7 +20,6 @@ import java.lang.reflect.Type;
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -28,6 +27,7 @@ import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeaders;
 
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
 import org.springframework.kafka.support.JacksonPresent;
@@ -54,7 +54,7 @@ import org.springframework.util.Assert;
  */
 public class MessagingMessageConverter implements RecordMessageConverter {
 
-	protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR
+	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); // NOSONAR
 
 	private boolean generateMessageId = false;
 
@@ -110,12 +110,10 @@ public class MessagingMessageConverter implements RecordMessageConverter {
 			this.headerMapper.toHeaders(record.headers(), rawHeaders);
 		}
 		else {
-			if (this.logger.isDebugEnabled()) {
-				this.logger.debug(
-						"No header mapper is available; Jackson is required for the default mapper; "
-						+ "headers (if present) are not mapped but provided raw in "
-						+ KafkaHeaders.NATIVE_HEADERS);
-			}
+			this.logger.debug(() ->
+					"No header mapper is available; Jackson is required for the default mapper; "
+					+ "headers (if present) are not mapped but provided raw in "
+					+ KafkaHeaders.NATIVE_HEADERS);
 			rawHeaders.put(KafkaHeaders.NATIVE_HEADERS, record.headers());
 		}
 		String ttName = record.timestampType() != null ? record.timestampType().name() : null;

spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java

@@ -38,7 +38,6 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -51,6 +50,7 @@ import org.junit.Test;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
@@ -74,7 +74,7 @@ import org.springframework.kafka.test.utils.KafkaTestUtils;
  */
 public class ConcurrentMessageListenerContainerTests {
 
-	private final Log logger = LogFactory.getLog(this.getClass());
+	private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass()));
 
 	private static String topic1 = "testTopic1";
 

spring-kafka/src/test/java/org/springframework/kafka/listener/FailedRecordTrackerTests.java

@@ -21,10 +21,11 @@ import static org.mockito.Mockito.mock;
 
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.apache.commons.logging.Log;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.core.log.LogAccessor;
+
 /**
  * @author Gary Russell
  * @since 2.2.5
@@ -37,7 +38,7 @@ public class FailedRecordTrackerTests {
 		AtomicBoolean recovered = new AtomicBoolean();
 		FailedRecordTracker tracker = new FailedRecordTracker((r, e) -> {
 			recovered.set(true);
-		}, 1, mock(Log.class));
+		}, 1, mock(LogAccessor.class));
 		ConsumerRecord<?, ?> record = new ConsumerRecord<>("foo", 0, 0L, "bar", "baz");
 		assertThat(tracker.skip(record, new RuntimeException())).isTrue();
 		assertThat(recovered.get()).isTrue();
@@ -48,7 +49,7 @@ public class FailedRecordTrackerTests {
 		AtomicBoolean recovered = new AtomicBoolean();
 		FailedRecordTracker tracker = new FailedRecordTracker((r, e) -> {
 			recovered.set(true);
-		}, 4, mock(Log.class));
+		}, 4, mock(LogAccessor.class));
 		ConsumerRecord<?, ?> record = new ConsumerRecord<>("foo", 0, 0L, "bar", "baz");
 		assertThat(tracker.skip(record, new RuntimeException())).isFalse();
 		assertThat(tracker.skip(record, new RuntimeException())).isFalse();
@@ -59,7 +60,7 @@ public class FailedRecordTrackerTests {
 
 	@Test
 	public void testSuccessAfterFailure() {
-		FailedRecordTracker tracker = new FailedRecordTracker(null, 2, mock(Log.class));
+		FailedRecordTracker tracker = new FailedRecordTracker(null, 2, mock(LogAccessor.class));
 		ConsumerRecord<?, ?> record = new ConsumerRecord<>("foo", 0, 0L, "bar", "baz");
 		assertThat(tracker.skip(record, new RuntimeException())).isFalse();
 		record = new ConsumerRecord<>("bar", 0, 0L, "bar", "baz");

spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

@@ -25,7 +25,6 @@ import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willAnswer;
-import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
@@ -50,10 +49,10 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -76,6 +75,7 @@ import org.mockito.InOrder;
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
@@ -110,7 +110,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
  */
 public class KafkaMessageListenerContainerTests {
 
-	private final Log logger = LogFactory.getLog(this.getClass());
+	private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass()));
 
 	private static String topic1 = "testTopic1";
 
@@ -1791,6 +1791,7 @@ public class KafkaMessageListenerContainerTests {
 		this.logger.info("Stop JSON4");
 	}
 
+	@SuppressWarnings({ "unchecked", "unchecked" })
 	@Test
 	public void testStaticAssign() throws Exception {
 		this.logger.info("Start static");
@@ -1813,9 +1814,12 @@ public class KafkaMessageListenerContainerTests {
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
 		container.setBeanName("testStatic");
-		Log consumerLogger = mock(Log.class);
-		given(consumerLogger.isDebugEnabled()).willReturn(true);
-		given(consumerLogger.isTraceEnabled()).willReturn(true);
+		LogAccessor consumerLogger = mock(LogAccessor.class);
+		List<String> log = new ArrayList<>();
+		willAnswer(inv -> {
+			log.add((String) ((Supplier<Object>) inv.getArgument(0)).get());
+			return null;
+		}).given(consumerLogger).trace(any(Supplier.class));
 		container.start();
 
 		ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
@@ -1831,9 +1835,7 @@ public class KafkaMessageListenerContainerTests {
 		container.stop();
 		pf.destroy();
 		this.logger.info("Stop static");
-		ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
-		verify(consumerLogger, atLeastOnce()).trace(captor.capture());
-		assertThat(captor.getAllValues()).contains("[testTopic22-0@0]");
+		assertThat(log).contains("[testTopic22-0@0]");
 	}
 
 	@Test

spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java

@@ -46,7 +46,6 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
-import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -66,6 +65,7 @@ import org.junit.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.InOrder;
 
+import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
@@ -100,7 +100,7 @@ import kafka.server.KafkaConfig;
  */
 public class TransactionalContainerTests {
 
-	private final Log logger = LogFactory.getLog(this.getClass());
+	private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass()));
 
 	private static String topic1 = "txTopic1";
 

spring-kafka/src/test/java/org/springframework/kafka/support/LogIfLevelEnabledTests.java

@@ -17,182 +17,132 @@
 package org.springframework.kafka.support;
 
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.BDDMockito.given;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.withSettings;
+
+import java.util.function.Supplier;
 
-import org.apache.commons.logging.Log;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.core.log.LogAccessor;
+
 /**
  * @author Gary Russell
  * @since 2.2.1
  *
  */
+@SuppressWarnings("unchecked")
 public class LogIfLevelEnabledTests {
 
 	private static final RuntimeException rte = new RuntimeException();
 
 	@Test
 	public void testFatalNoEx() {
-		Log theLogger = mock(Log.class);
+		LogAccessor theLogger = mock(LogAccessor.class);
 		LogIfLevelEnabled logger = new LogIfLevelEnabled(theLogger, LogIfLevelEnabled.Level.FATAL);
-		given(theLogger.isFatalEnabled()).willReturn(true);
 		logger.log(() -> "foo");
-		verify(theLogger).isFatalEnabled();
-		verify(theLogger).fatal(any());
+		verify(theLogger).fatal(any(Supplier.class));
 		verifyNoMoreInteractions(theLogger);
 	}
 
 	@Test
 	public void testErrorNoEx() {
-		Log theLogger = mock(Log.class);
+		LogAccessor theLogger = mock(LogAccessor.class);
 		LogIfLevelEnabled logger = new LogIfLevelEnabled(theLogger, LogIfLevelEnabled.Level.ERROR);
-		given(theLogger.isFatalEnabled()).willReturn(true);
-		given(theLogger.isErrorEnabled()).willReturn(true);
 		logger.log(() -> "foo");
-		verify(theLogger).isErrorEnabled();
-		verify(theLogger).error(any());
+		verify(theLogger).error(any(Supplier.class));
 		verifyNoMoreInteractions(theLogger);
 	}
 
 	@Test
 	public void testWarnNoEx() {
-		Log theLogger = mock(Log.class);
+		LogAccessor theLogger = mock(LogAccessor.class);
 		LogIfLevelEnabled logger = new LogIfLevelEnabled(theLogger, LogIfLevelEnabled.Level.WARN);
-		given(theLogger.isFatalEnabled()).willReturn(true);
-		given(theLogger.isErrorEnabled()).willReturn(true);
-		given(theLogger.isWarnEnabled()).willReturn(true);
 		logger.log(() -> "foo");
-		verify(theLogger).isWarnEnabled();
-		verify(theLogger).warn(any());
+		verify(theLogger).warn(any(Supplier.class));
 		verifyNoMoreInteractions(theLogger);
 	}
 
 	@Test
 	public void testInfoNoEx() {
-		Log theLogger = mock(Log.class);
+		LogAccessor theLogger = mock(LogAccessor.class);
 		LogIfLevelEnabled logger = new LogIfLevelEnabled(theLogger, LogIfLevelEnabled.Level.INFO);
-		given(theLogger.isFatalEnabled()).willReturn(true);
-		given(theLogger.isErrorEnabled()).willReturn(true);
-		given(theLogger.isWarnEnabled()).willReturn(true);
-		given(theLogger.isInfoEnabled()).willReturn(true);
 		logger.log(() -> "foo");
-		verify(theLogger).isInfoEnabled();
-		verify(theLogger).info(any());
+		verify(theLogger).info(any(Supplier.class));
 		verifyNoMoreInteractions(theLogger);
 	}
 
 	@Test
 	public void testDebugNoEx() {
-		Log theLogger = mock(Log.class);
+		LogAccessor theLogger = mock(LogAccessor.class);
 		LogIfLevelEnabled logger = new LogIfLevelEnabled(theLogger, LogIfLevelEnabled.Level.DEBUG);
-		given(theLogger.isFatalEnabled()).willReturn(true);
-		given(theLogger.isErrorEnabled()).willReturn(true);
-		given(theLogger.isWarnEnabled()).willReturn(true);
-		given(theLogger.isInfoEnabled()).willReturn(true);
-		given(theLogger.isDebugEnabled()).willReturn(true);
 		logger.log(() -> "foo");
-		verify(theLogger).isDebugEnabled();
-		verify(theLogger).debug(any());
+		verify(theLogger).debug(any(Supplier.class));
 		verifyNoMoreInteractions(theLogger);
 	}
 
 	@Test
 	public void testTraceNoEx() {
-		Log theLogger = mock(Log.class);
+		LogAccessor theLogger = mock(LogAccessor.class);
 		LogIfLevelEnabled logger = new LogIfLevelEnabled(theLogger, LogIfLevelEnabled.Level.TRACE);
-		given(theLogger.isFatalEnabled()).willReturn(true);
-		given(theLogger.isErrorEnabled()).willReturn(true);
-		given(theLogger.isWarnEnabled()).willReturn(true);
-		given(theLogger.isInfoEnabled()).willReturn(true);
-		given(theLogger.isDebugEnabled()).willReturn(true);
-		given(theLogger.isTraceEnabled()).willReturn(true);
 		logger.log(() -> "foo");
-		verify(theLogger).isTraceEnabled();
-		verify(theLogger).trace(any());
+		verify(theLogger).trace(any(Supplier.class));
 		verifyNoMoreInteractions(theLogger);
 	}
 
 	@Test
 	public void testFatalWithEx() {
-		Log theLogger = mock(Log.class);
+		LogAccessor theLogger = mock(LogAccessor.class);
 		LogIfLevelEnabled logger = new LogIfLevelEnabled(theLogger, LogIfLevelEnabled.Level.FATAL);
-		given(theLogger.isFatalEnabled()).willReturn(true);
 		logger.log(() -> "foo", rte);
-		verify(theLogger).isFatalEnabled();
-		verify(theLogger).fatal(any(), any());
+		verify(theLogger).fatal(any(), any(Supplier.class));
 		verifyNoMoreInteractions(theLogger);
 	}
 
 	@Test
 	public void testErrorWithEx() {
-		Log theLogger = mock(Log.class);
+		LogAccessor theLogger = mock(LogAccessor.class);
 		LogIfLevelEnabled logger = new LogIfLevelEnabled(theLogger, LogIfLevelEnabled.Level.ERROR);
-		given(theLogger.isFatalEnabled()).willReturn(true);
-		given(theLogger.isErrorEnabled()).willReturn(true);
 		logger.log(() -> "foo", rte);
-		verify(theLogger).isErrorEnabled();
-		verify(theLogger).error(any(), any());
+		verify(theLogger).error(any(), any(Supplier.class));
 		verifyNoMoreInteractions(theLogger);
 	}
 
 	@Test
 	public void testWarnWithEx() {
-		Log theLogger = mock(Log.class);
+		LogAccessor theLogger = mock(LogAccessor.class);
 		LogIfLevelEnabled logger = new LogIfLevelEnabled(theLogger, LogIfLevelEnabled.Level.WARN);
-		given(theLogger.isFatalEnabled()).willReturn(true);
-		given(theLogger.isErrorEnabled()).willReturn(true);
-		given(theLogger.isWarnEnabled()).willReturn(true);
 		logger.log(() -> "foo", rte);
-		verify(theLogger).isWarnEnabled();
-		verify(theLogger).warn(any(), any());
+		verify(theLogger).warn(any(), any(Supplier.class));
 		verifyNoMoreInteractions(theLogger);
 	}
 
 	@Test
 	public void testInfoWithEx() {
-		Log theLogger = mock(Log.class);
+		LogAccessor theLogger = mock(LogAccessor.class);
 		LogIfLevelEnabled logger = new LogIfLevelEnabled(theLogger, LogIfLevelEnabled.Level.INFO);
-		given(theLogger.isFatalEnabled()).willReturn(true);
-		given(theLogger.isErrorEnabled()).willReturn(true);
-		given(theLogger.isWarnEnabled()).willReturn(true);
-		given(theLogger.isInfoEnabled()).willReturn(true);
 		logger.log(() -> "foo", rte);
-		verify(theLogger).isInfoEnabled();
-		verify(theLogger).info(any(), any());
+		verify(theLogger).info(any(), any(Supplier.class));
 		verifyNoMoreInteractions(theLogger);
 	}
 
 	@Test
 	public void testDebugWithEx() {
-		Log theLogger = mock(Log.class);
+		LogAccessor theLogger = mock(LogAccessor.class, withSettings().verboseLogging());
 		LogIfLevelEnabled logger = new LogIfLevelEnabled(theLogger, LogIfLevelEnabled.Level.DEBUG);
-		given(theLogger.isFatalEnabled()).willReturn(true);
-		given(theLogger.isErrorEnabled()).willReturn(true);
-		given(theLogger.isWarnEnabled()).willReturn(true);
-		given(theLogger.isInfoEnabled()).willReturn(true);
-		given(theLogger.isDebugEnabled()).willReturn(true);
 		logger.log(() -> "foo", rte);
-		verify(theLogger).isDebugEnabled();
-		verify(theLogger).debug(any(), any());
+		verify(theLogger).debug(any(), any(Supplier.class));
 		verifyNoMoreInteractions(theLogger);
 	}
 
 	@Test
 	public void testTraceWithEx() {
-		Log theLogger = mock(Log.class);
+		LogAccessor theLogger = mock(LogAccessor.class);
 		LogIfLevelEnabled logger = new LogIfLevelEnabled(theLogger, LogIfLevelEnabled.Level.TRACE);
-		given(theLogger.isFatalEnabled()).willReturn(true);
-		given(theLogger.isErrorEnabled()).willReturn(true);
-		given(theLogger.isWarnEnabled()).willReturn(true);
-		given(theLogger.isInfoEnabled()).willReturn(true);
-		given(theLogger.isDebugEnabled()).willReturn(true);
-		given(theLogger.isTraceEnabled()).willReturn(true);
 		logger.log(() -> "foo", rte);
-		verify(theLogger).isTraceEnabled();
-		verify(theLogger).trace(any(), any());
+		verify(theLogger).trace(any(), any(Supplier.class));
 		verifyNoMoreInteractions(theLogger);
 	}