Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-537: Configurable log level for offset commits

Browse Source 
Resolves https://github.com/spring-projects/spring-kafka/issues/537

Use a `Supplier` to avoid early string concatenation.

Docs

Polishing
This commit is contained in:
Gary Russell
2018-01-28 15:39:56 -05:00
committed by Artem Bilan
parent 63b2bcb018
commit b1781766d2
6 changed files with 205 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<K, V> extends AbstractMessageListener
private final ScheduledFuture<?> monitorTask;
private final LogIfLevelEnabled commitLogger = new LogIfLevelEnabled(this.logger,
this.containerProperties.getCommitLogLevel());
private volatile Map<TopicPartition, OffsetMetadata> definedPartitions;
private volatile Collection<TopicPartition> assignedPartitions;
@@ -529,9 +533,7 @@ public class KafkaMessageListenerContainer<K, V> 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<K, V> extends AbstractMessageListener
Map<TopicPartition, OffsetAndMetadata> 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<K, V> 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<K, V> extends AbstractMessageListener
Map<TopicPartition, OffsetAndMetadata> 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<K, V> extends AbstractMessageListener
private void sendOffsetsToTransaction(Producer producer) {
handleAcks();
Map<TopicPartition, OffsetAndMetadata> commits = buildCommits();
this.commitLogger.log(() -> "Sending offsets to transaction: " + commits);
producer.sendOffsetsToTransaction(commits, this.consumerGroupId);
}
@@ -1209,9 +1211,7 @@ public class KafkaMessageListenerContainer<K, V> 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);

27
spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java
View File

@@ -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 ["

153
spring-kafka/src/main/java/org/springframework/kafka/support/LogIfLevelEnabled.java Normal file
View File

@@ -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<Object> 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<Object> 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;
}
}
}

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

@@ -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<Integer, String> container = spyOnContainer(

5
src/reference/asciidoc/kafka.adoc
View File

@@ -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:

7
src/reference/asciidoc/whats-new.adoc
View File

@@ -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 <<kafka-container>> 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].