From b1781766d220f83a78d0cfb120e1da7d8328a503 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sun, 28 Jan 2018 15:39:56 -0500 Subject: [PATCH] GH-537: Configurable log level for offset commits Resolves https://github.com/spring-projects/spring-kafka/issues/537 Use a `Supplier` to avoid early string concatenation. Docs Polishing --- .../KafkaMessageListenerContainer.java | 26 +-- .../listener/config/ContainerProperties.java | 27 +++- .../kafka/support/LogIfLevelEnabled.java | 153 ++++++++++++++++++ .../KafkaMessageListenerContainerTests.java | 3 +- src/reference/asciidoc/kafka.adoc | 5 + src/reference/asciidoc/whats-new.adoc | 7 + 6 files changed, 205 insertions(+), 16 deletions(-) create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/support/LogIfLevelEnabled.java 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 6de21f52..9c691f31 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 @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -56,6 +56,7 @@ import org.springframework.kafka.event.NonResponsiveConsumerEvent; import org.springframework.kafka.listener.ConsumerSeekAware.ConsumerSeekCallback; import org.springframework.kafka.listener.config.ContainerProperties; import org.springframework.kafka.support.Acknowledgment; +import org.springframework.kafka.support.LogIfLevelEnabled; import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.kafka.support.TopicPartitionInitialOffset.SeekPosition; import org.springframework.kafka.transaction.KafkaTransactionManager; @@ -371,6 +372,9 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private final ScheduledFuture monitorTask; + private final LogIfLevelEnabled commitLogger = new LogIfLevelEnabled(this.logger, + this.containerProperties.getCommitLogLevel()); + private volatile Map definedPartitions; private volatile Collection assignedPartitions; @@ -529,9 +533,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener return; } } - if (ListenerConsumer.this.logger.isDebugEnabled()) { - ListenerConsumer.this.logger.debug("Committing on assignment: " + offsets); - } + ListenerConsumer.this.commitLogger.log(() -> "Committing on assignment: " + offsets); if (ListenerConsumer.this.transactionTemplate != null && ListenerConsumer.this.kafkaTxManager != null) { ListenerConsumer.this.transactionTemplate.execute(new TransactionCallbackWithoutResult() { @@ -755,15 +757,12 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener Map commits = Collections.singletonMap( new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1)); - if (ListenerConsumer.this.logger.isDebugEnabled()) { - ListenerConsumer.this.logger.debug("Committing: " + commits); - } + this.commitLogger.log(() -> "Committing: " + commits); if (this.containerProperties.isSyncCommits()) { - ListenerConsumer.this.consumer.commitSync(commits); + this.consumer.commitSync(commits); } else { - ListenerConsumer.this.consumer.commitAsync(commits, - ListenerConsumer.this.commitCallback); + this.consumer.commitAsync(commits, this.commitCallback); } } @@ -995,6 +994,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener Collections.singletonMap(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1)); if (producer == null) { + this.commitLogger.log(() -> "Committing: " + offsetsToCommit); if (this.containerProperties.isSyncCommits()) { this.consumer.commitSync(offsetsToCommit); } @@ -1019,6 +1019,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener Map offsetsToCommit = Collections.singletonMap(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1)); + this.commitLogger.log(() -> "Committing: " + offsetsToCommit); if (this.containerProperties.isSyncCommits()) { this.consumer.commitSync(offsetsToCommit); } @@ -1072,6 +1073,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private void sendOffsetsToTransaction(Producer producer) { handleAcks(); Map commits = buildCommits(); + this.commitLogger.log(() -> "Sending offsets to transaction: " + commits); producer.sendOffsetsToTransaction(commits, this.consumerGroupId); } @@ -1209,9 +1211,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener this.logger.debug("Commit list: " + commits); } if (!commits.isEmpty()) { - if (this.logger.isDebugEnabled()) { - this.logger.debug("Committing: " + commits); - } + this.commitLogger.log(() -> "Committing: " + commits); try { if (this.containerProperties.isSyncCommits()) { this.consumer.commitSync(commits); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java index 1d9362df..1f02a691 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -29,6 +29,7 @@ import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMo import org.springframework.kafka.listener.BatchErrorHandler; import org.springframework.kafka.listener.ErrorHandler; import org.springframework.kafka.listener.GenericErrorHandler; +import org.springframework.kafka.support.LogIfLevelEnabled; import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.scheduling.TaskScheduler; import org.springframework.transaction.PlatformTransactionManager; @@ -162,6 +163,8 @@ public class ContainerProperties { private boolean logContainerConfig; + private LogIfLevelEnabled.Level commitLogLevel = LogIfLevelEnabled.Level.DEBUG; + public ContainerProperties(String... topics) { Assert.notEmpty(topics, "An array of topicPartitions must be provided"); this.topics = Arrays.asList(topics).toArray(new String[topics.length]); @@ -490,7 +493,7 @@ public class ContainerProperties { /** * Log the container configuration if true (INFO). * @return true to log. - * @since 2.0.1 + * @since 2.1.1 */ public boolean isLogContainerConfig() { return this.logContainerConfig; @@ -505,6 +508,26 @@ public class ContainerProperties { this.logContainerConfig = logContainerConfig; } + /** + * The level at which to log offset commits. + * @return the level. + * @since 2.1.2 + */ + public LogIfLevelEnabled.Level getCommitLogLevel() { + return this.commitLogLevel; + } + + /** + * Set the level at which to log offset commits. + * Default: DEBUG. + * @param commitLogLevel the level. + * @since 2.1.2 + */ + public void setCommitLogLevel(LogIfLevelEnabled.Level commitLogLevel) { + Assert.notNull(commitLogLevel, "'commitLogLevel' cannot be nul"); + this.commitLogLevel = commitLogLevel; + } + @Override public String toString() { return "ContainerProperties [" 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 new file mode 100644 index 00000000..c7899852 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/LogIfLevelEnabled.java @@ -0,0 +1,153 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.support; + +import java.util.function.Supplier; + +import org.apache.commons.logging.Log; + +import org.springframework.util.Assert; + +/** + * Wrapper for a commons-logging Log supporting configurable + * logging levels. + * + * @author Gary Russell + * @since 2.1.2 + * + */ +public final class LogIfLevelEnabled { + + private final Log logger; + + private final Level level; + + public LogIfLevelEnabled(Log logger, Level level) { + Assert.notNull(logger, "'logger' cannot be null"); + Assert.notNull(level, "'level' cannot be null"); + this.logger = logger; + this.level = level; + } + + /** + * Logging levels. + */ + public enum Level { + + /** + * Fatal. + */ + FATAL, + + /** + * Error. + */ + ERROR, + + /** + * Warn. + */ + WARN, + + /** + * Info. + */ + INFO, + + /** + * Debug. + */ + DEBUG, + + /** + * Trace. + */ + TRACE + + } + + public void log(Supplier messageSupplier) { + switch (this.level) { + case FATAL: + if (this.logger.isFatalEnabled()) { + this.logger.fatal(messageSupplier.get()); + } + break; + case ERROR: + if (this.logger.isErrorEnabled()) { + this.logger.error(messageSupplier.get()); + } + break; + case WARN: + if (this.logger.isWarnEnabled()) { + this.logger.warn(messageSupplier.get()); + } + break; + case INFO: + if (this.logger.isInfoEnabled()) { + this.logger.info(messageSupplier.get()); + } + break; + case DEBUG: + if (this.logger.isDebugEnabled()) { + this.logger.debug(messageSupplier.get()); + } + break; + case TRACE: + if (this.logger.isTraceEnabled()) { + this.logger.trace(messageSupplier.get()); + } + break; + } + } + + public void log(Supplier messageSupplier, Throwable t) { + switch (this.level) { + case FATAL: + if (this.logger.isFatalEnabled()) { + this.logger.fatal(messageSupplier.get(), t); + } + break; + case ERROR: + if (this.logger.isErrorEnabled()) { + this.logger.error(messageSupplier.get(), t); + } + break; + case WARN: + if (this.logger.isWarnEnabled()) { + this.logger.warn(messageSupplier.get(), t); + } + break; + case INFO: + if (this.logger.isInfoEnabled()) { + this.logger.info(messageSupplier.get(), t); + } + break; + case DEBUG: + if (this.logger.isDebugEnabled()) { + this.logger.debug(messageSupplier.get(), t); + } + break; + case TRACE: + if (this.logger.isTraceEnabled()) { + this.logger.trace(messageSupplier.get(), t); + } + break; + } + } + +} 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 f4d2907b..f8d2a310 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 @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -441,6 +441,7 @@ public class KafkaMessageListenerContainerTests { containerProps.setSyncCommits(true); containerProps.setAckMode(AckMode.RECORD); containerProps.setAckOnError(false); +// containerProps.setCommitLogLevel(LogIfLevelEnabled.Level.WARN); CountDownLatch stubbingComplete = new CountDownLatch(1); KafkaMessageListenerContainer container = spyOnContainer( diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 99953e19..f57e71f4 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -441,6 +441,7 @@ The `KafkaMessageListenerContainer` receives all message from all topics/partiti The `ConcurrentMessageListenerContainer` delegates to 1 or more `KafkaMessageListenerContainer` s to provide multi-threaded consumption. +[[kafka-container]] ====== KafkaMessageListenerContainer The following constructors are available. @@ -498,6 +499,10 @@ Refer to the JavaDocs for `ContainerProperties` for more information about the v Since version _2.1.1_, a new property `logContainerConfig` is available; when true, and INFO logging is enabled, each listener container will write a log message summarizing its configuration properties. +By default, logging of topic offset commits is performed with the DEBUG logging level. +Starting with _version 2.1.2_, there is a new property in `ContainerProperties` called `commitLogLevel` which allows you to specify the log level for these messages. +For example, to change the log level to INFO, use `containerProperties.setCommitLogLevel(LogIfLevelEnabled.Level.INFO);`. + ====== ConcurrentMessageListenerContainer The single constructor is similar to the first `KafkaListenerContainer` constructor: diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index bdb59f6d..93d7aef4 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -22,6 +22,13 @@ Starting with _version 2.1.1_, it is now possible to set the `client.id` prefix Previously, to customize the client id, you would need a separate consumer factory (and container factory) per listener. The prefix is suffixed with `-n` to provide unique client ids when using concurrency. + +==== Logging Offset Commits + +By default, logging of topic offset commits is performed with the DEBUG logging level. +Starting with _version 2.1.2_, there is a new property in `ContainerProperties` called `commitLogLevel` which allows you to specify the log level for these messages. +See <> for more information. + ==== Migration Guide from 2.0 https://github.com/spring-projects/spring-kafka/wiki/Spring-for-Apache-Kafka-2.0-to-2.1-Migration-Guide[2.0 to 2.1 Migration].