diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java index b3d0d071..591dcc8f 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java +++ b/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 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()) diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java index b77cde80..fc5610cf 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java +++ b/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 ConsumerRecords getRecords(Consumer consumer, long timeout) { logger.debug("Polling..."); ConsumerRecords 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; } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java index 01953051..56401733 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java +++ b/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 private final Set> 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 } 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 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 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..."); } } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java index 24a5877e..3c39063a 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java +++ b/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, K, V> implements KafkaListenerContainerFactory, 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); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java index 703e0c4d..eb92662c 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java +++ b/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 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 if (this.recordFilterStrategy != null) { if (this.batchListener) { if (((MessagingMessageListenerAdapter) 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<>( diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java b/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java index 07f879aa..200e0bb1 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java +++ b/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 { - protected final Log logger = LogFactory.getLog(getClass()); //NOSONAR + protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); //NOSONAR private final Map listenerContainers = new ConcurrentHashMap(); @@ -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"); } } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/MethodKafkaListenerEndpoint.java b/spring-kafka/src/main/java/org/springframework/kafka/config/MethodKafkaListenerEndpoint.java index 440026b1..c14aed6f 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/MethodKafkaListenerEndpoint.java +++ b/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 extends AbstractKafkaListenerEndpoint { - 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 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) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java b/spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java index b6b3588f..4dad8b46 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java +++ b/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 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 implements ProducerFactory, */ 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 configs; @@ -145,11 +145,9 @@ public class DefaultKafkaProducerFactory implements ProducerFactory, 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 implements ProducerFactory, */ 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 implements ProducerFactory, 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 implements ProducerFactory, * @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 implements ProducerFactory, @Override public Future send(ProducerRecord record) { + LOGGER.trace(() -> toString() + " send(" + record + ")"); return this.delegate.send(record); } @Override public Future send(ProducerRecord 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 implements ProducerFactory, @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 implements ProducerFactory, public void sendOffsetsToTransaction(Map 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 implements ProducerFactory, @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 implements ProducerFactory, @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(); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java index 136b96c2..46573498 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java +++ b/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 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 } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java index 91f92d28..b974b55d 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java +++ b/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 implements KafkaOperations { - protected final Log logger = LogFactory.getLog(this.getClass()); //NOSONAR + protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass())); //NOSONAR private final ProducerFactory producerFactory; @@ -370,17 +370,13 @@ public class KafkaTemplate implements KafkaOperations { + "run in a transaction started by a listener container when consuming a record"); } final Producer producer = getTheProducer(); - if (this.logger.isTraceEnabled()) { - this.logger.trace("Sending: " + producerRecord); - } + this.logger.trace(() -> "Sending: " + producerRecord); final SettableListenableFuture> 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 implements KafkaOperations { 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 { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java index 51f85fe5..b328b18e 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java +++ b/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 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 consumerFactory; // NOSONAR (final) @@ -330,7 +330,7 @@ public abstract class AbstractMessageListenerContainer } } 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 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 @Override public void onPartitionsRevoked(Collection 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 partitions) { - Log logger2 = AbstractMessageListenerContainer.this.logger; - if (logger2.isInfoEnabled()) { - logger2.info(getGroupId() + ": partitions assigned: " + partitions); - } + AbstractMessageListenerContainer.this.logger.info(() -> + getGroupId() + ": partitions assigned: " + partitions); } }; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/BatchLoggingErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/BatchLoggingErrorHandler.java index 5c267e48..9b477c59 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/BatchLoggingErrorHandler.java +++ b/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)); } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java index f7562f8f..da7c4b02 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java @@ -137,7 +137,7 @@ public class ConcurrentMessageListenerContainer 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; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java index 07bfbacc..e52ecac5 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java +++ b/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, 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, Exception, TopicPartition> DEFAULT_DESTINATION_RESOLVER = (cr, e) -> new TopicPartition(cr.topic() + ".DLT", cr.partition()); @@ -140,10 +141,10 @@ public class DeadLetterPublishingRecoverer implements BiConsumer outRecord = createProducerRecord(record, tp, headers, deserEx == null ? null : deserEx.getData()); @@ -171,9 +172,7 @@ public class DeadLetterPublishingRecoverer implements BiConsumer) 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) this.templates.values() .stream() .reduce((first, second) -> second) @@ -211,15 +210,13 @@ public class DeadLetterPublishingRecoverer implements BiConsumer outRecord, KafkaOperations 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); } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessor.java index 9c657cc6..34921427 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessor.java +++ b/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 implements AfterRollbackProcessor { - 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 implements AfterRollbackProcess */ public DefaultAfterRollbackProcessor(@Nullable BiConsumer, 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 implements AfterRollbackProcess public void process(List> records, Consumer 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 skipped = records.get(0); this.kafkaTemplate.sendOffsetsToTransaction( diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordTracker.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordTracker.java index a983702c..77cc00ed 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordTracker.java +++ b/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, Exception> recoverer, int maxFailures, Log logger) { + FailedRecordTracker(@Nullable BiConsumer, 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; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index f6e57587..44478392 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/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 // 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 // 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 // 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 props = KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties(); this.checkNullKeyForExceptions = checkDeserializer(findDeserializerClass(props, false)); @@ -613,8 +613,10 @@ public class KafkaMessageListenerContainer // 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 // 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 // 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 // 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 // NOSONAR line count } private void debugRecords(ConsumerRecords 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 // 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 paused = this.consumer.paused(); this.consumer.resume(paused); this.consumerPaused = false; @@ -891,9 +889,7 @@ public class KafkaMessageListenerContainer // 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 // 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 // NOSONAR line count private void handleAcks() { ConsumerRecord 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 record) { + this.logger.trace(() -> "Ack: " + record); + } + private void processAck(ConsumerRecord record) { if (!Thread.currentThread().equals(this.consumerThread)) { try { @@ -1028,7 +1026,7 @@ public class KafkaMessageListenerContainer // NOSONAR line count }); } catch (RuntimeException e) { - this.logger.error("Transaction rolled back", e); + this.logger.error(e, "Transaction rolled back"); AfterRollbackProcessor afterRollbackProcessorToUse = (AfterRollbackProcessor) getAfterRollbackProcessor(); if (afterRollbackProcessorToUse.isProcessInTransaction() && this.transactionTemplate != null) { @@ -1090,11 +1088,11 @@ public class KafkaMessageListenerContainer // 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 // NOSONAR line count Iterator> iterator = records.iterator(); while (iterator.hasNext()) { final ConsumerRecord 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 // 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 // NOSONAR line count Iterator> iterator = records.iterator(); while (iterator.hasNext()) { final ConsumerRecord 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 // 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 // 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 // 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 // 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 // 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 // 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 // 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 record = this.acks.poll(); while (record != null) { @@ -1569,9 +1565,7 @@ public class KafkaMessageListenerContainer // NOSONAR line count private void commitIfNecessary() { Map 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 // 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 // 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 // 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 // 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(); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java index c455e7ca..d48e3ff1 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java +++ b/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; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/LoggingCommitCallback.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/LoggingCommitCallback.java index 3593055e..82cd376a 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/LoggingCommitCallback.java +++ b/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 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"); } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/LoggingErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/LoggingErrorHandler.java index 0d47e4bb..41e18788 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/LoggingErrorHandler.java +++ b/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)); } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java index 1e064b0a..9cc61c01 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java +++ b/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, 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()); } } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractDelegatingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractDelegatingMessageListenerAdapter.java index 5b2abb92..6616de12 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AbstractDelegatingMessageListenerAdapter.java +++ b/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 implements ConsumerSeekAware, DelegatingMessageListener { - 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 diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java index a2b070d9..7aa9577f 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java @@ -132,9 +132,7 @@ public class BatchMessagingMessageListenerAdapter 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); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java index 42c8e8ea..14664415 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java +++ b/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 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 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 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 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; } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordMessagingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordMessagingMessageListenerAdapter.java index c8a4b141..714f245a 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordMessagingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordMessagingMessageListenerAdapter.java @@ -72,9 +72,7 @@ public class RecordMessagingMessageListenerAdapter extends MessagingMessag @Override public void onMessage(ConsumerRecord 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) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/requestreply/AggregatingReplyingKafkaTemplate.java b/spring-kafka/src/main/java/org/springframework/kafka/requestreply/AggregatingReplyingKafkaTemplate.java index 2bf8a27c..df0c0438 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/requestreply/AggregatingReplyingKafkaTemplate.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/requestreply/AggregatingReplyingKafkaTemplate.java @@ -122,7 +122,7 @@ public class AggregatingReplyingKafkaTemplate 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"); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java b/spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java index 5ca2dbc8..0eea6c04 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java @@ -121,7 +121,7 @@ public class ReplyingKafkaTemplate extends KafkaTemplate 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 extends KafkaTemplate 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 future = new RequestReplyFuture<>(); this.futures.put(correlationId, future); try { @@ -275,9 +273,7 @@ public class ReplyingKafkaTemplate extends KafkaTemplate implemen this.scheduler.schedule(() -> { RequestReplyFuture 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 extends KafkaTemplate 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 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 extends KafkaTemplate implemen protected void logLateArrival(ConsumerRecord 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)); } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java index 138208a9..0ef109ac 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java +++ b/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 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; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java index e64d9e05..613fdf97 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java +++ b/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; } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/LogIfLevelEnabled.java b/spring-kafka/src/main/java/org/springframework/kafka/support/LogIfLevelEnabled.java index 2a105290..dd87c4e9 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/LogIfLevelEnabled.java +++ b/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 messageSupplier) { + public void log(Supplier messageSupplier) { switch (this.level) { case FATAL: fatal(messageSupplier, null); @@ -103,92 +102,80 @@ public final class LogIfLevelEnabled { } } - public void log(Supplier messageSupplier, Throwable t) { + public void log(Supplier 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 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 messageSupplier, Throwable thrown) { + if (thrown != null) { + this.logger.fatal(thrown, messageSupplier); + } + else { + this.logger.fatal(messageSupplier); } } - private void error(Supplier 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 messageSupplier, Throwable thrown) { + if (thrown != null) { + this.logger.error(thrown, messageSupplier); + } + else { + this.logger.error(messageSupplier); } } - private void warn(Supplier 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 messageSupplier, Throwable thrown) { + if (thrown != null) { + this.logger.warn(thrown, messageSupplier); + } + else { + this.logger.warn(messageSupplier); } } - private void info(Supplier 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 messageSupplier, Throwable thrown) { + if (thrown != null) { + this.logger.info(thrown, messageSupplier); + } + else { + this.logger.info(messageSupplier); } } - private void debug(Supplier 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 messageSupplier, Throwable thrown) { + if (thrown != null) { + this.logger.debug(thrown, messageSupplier); + } + else { + this.logger.debug(messageSupplier); } } - private void trace(Supplier 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 messageSupplier, Throwable thrown) { + if (thrown != null) { + this.logger.trace(thrown, messageSupplier); + } + else { + this.logger.trace(messageSupplier); } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/LoggingProducerListener.java b/spring-kafka/src/main/java/org/springframework/kafka/support/LoggingProducerListener.java index b41f6286..16ca3853 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/LoggingProducerListener.java +++ b/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 implements ProducerListener { */ 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 implements ProducerListener { @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 implements ProducerListener { logOutput.append(" and partition ").append(partition); } logOutput.append(":"); - logger.error(logOutput, exception); - } + return logOutput.toString(); + }); } private String toDisplayString(String original, int maxCharacters) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/SeekUtils.java b/spring-kafka/src/main/java/org/springframework/kafka/support/SeekUtils.java index 47071858..4b4fe288 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/SeekUtils.java +++ b/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> records, Consumer consumer, Exception exception, - boolean recoverable, BiPredicate, Exception> skipper, Log logger) { + boolean recoverable, BiPredicate, Exception> skipper, LogAccessor logger) { Map 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(); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java index 558e2876..029aa79b 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java +++ b/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); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java index 5cd675e1..a12fd4a1 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java +++ b/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; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java index 77b3d035..90087e0a 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java +++ b/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"; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/FailedRecordTrackerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/FailedRecordTrackerTests.java index 717c5013..6e799581 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/FailedRecordTrackerTests.java +++ b/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"); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java index 9daa2aad..1922dae7 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java +++ b/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 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 log = new ArrayList<>(); + willAnswer(inv -> { + log.add((String) ((Supplier) 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 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 diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java index 083e01bf..dee48136 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java +++ b/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"; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/support/LogIfLevelEnabledTests.java b/spring-kafka/src/test/java/org/springframework/kafka/support/LogIfLevelEnabledTests.java index 008d6709..597f1996 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/support/LogIfLevelEnabledTests.java +++ b/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); }