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

GH-1499: Option to suppress ConsumerRecord logging

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

**I will do the backports because I expect many conflicts.**
This commit is contained in:
Gary Russell
2020-06-08 15:46:26 -04:00
parent 423deb0182
commit 3294f6c058
6 changed files with 63 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -222,6 +222,8 @@ public class ContainerProperties {
private Properties consumerProperties;
private boolean onlyLogRecordMetadata;
/**
* Create properties for a container that will subscribe to the specified topics.
* @param topics the topics.
@@ -646,6 +648,20 @@ public class ContainerProperties {
this.consumerProperties = consumerProperties;
}
public boolean isOnlyLogRecordMetadata() {
return this.onlyLogRecordMetadata;
}
/**
* Set to true to only log {@code topic-partition@offset} in log messages instead
* of {@code record.toString()}.
* @param onlyLogRecordMetadata true to only log the topic/parrtition/offset.
* @since 2.2.14
*/
public void setOnlyLogRecordMetadata(boolean onlyLogRecordMetadata) {
this.onlyLogRecordMetadata = onlyLogRecordMetadata;
}
@Override
public String toString() {
return "ContainerProperties ["

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

@@ -48,7 +48,8 @@ class FailedRecordTracker {
FailedRecordTracker(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, int maxFailures, Log logger) {
if (recoverer == null) {
this.recoverer = (r, t) -> logger.error("Max failures (" + maxFailures + ") reached for: " + r, t);
this.recoverer = (r, t) -> logger.error("Max failures (" + maxFailures + ") reached for: "
+ ListenerUtils.recordToString(r), t);
}
else {
this.recoverer = recoverer;

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

@@ -692,6 +692,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
@Override
public void run() {
ListenerUtils.setLogOnlyMetadata(this.containerProperties.isOnlyLogRecordMetadata());
this.consumerThread = Thread.currentThread();
if (this.genericListener instanceof ConsumerSeekAware) {
((ConsumerSeekAware) this.genericListener).registerSeekCallback(this);
@@ -911,7 +912,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
ConsumerRecord<K, V> record = this.acks.poll();
while (record != null) {
if (this.logger.isTraceEnabled()) {
this.logger.trace("Ack: " + record);
this.logger.trace("Ack: " + ListenerUtils.recordToString(record));
}
processAck(record);
record = this.acks.poll();
@@ -1172,7 +1173,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
while (iterator.hasNext()) {
final ConsumerRecord<K, V> record = iterator.next();
if (this.logger.isTraceEnabled()) {
this.logger.trace("Processing " + record);
this.logger.trace("Processing " + ListenerUtils.recordToString(record));
}
try {
TransactionSupport
@@ -1242,7 +1243,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
while (iterator.hasNext()) {
final ConsumerRecord<K, V> record = iterator.next();
if (this.logger.isTraceEnabled()) {
this.logger.trace("Processing " + record);
this.logger.trace("Processing " + ListenerUtils.recordToString(record));
}
doInvokeRecordListener(record, null, iterator);
}
@@ -1316,7 +1317,8 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
}
if (record == null) {
if (this.logger.isDebugEnabled()) {
this.logger.debug("RecordInterceptor returned null, skipping: " + recordArg);
this.logger.debug("RecordInterceptor returned null, skipping: "
+ ListenerUtils.recordToString(recordArg));
}
}
else {

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

@@ -16,6 +16,8 @@
package org.springframework.kafka.listener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.util.Assert;
/**
@@ -31,6 +33,8 @@ public final class ListenerUtils {
super();
}
private static final ThreadLocal<Boolean> LOG_METADATA_ONLY = new ThreadLocal<>();
public static ListenerType determineListenerType(Object listener) {
Assert.notNull(listener, "Listener cannot be null");
ListenerType listenerType;
@@ -55,4 +59,31 @@ public final class ListenerUtils {
return listenerType;
}
/**
* Set to true to only log record metadata.
* @param onlyMeta true to only log record metadata.
* @since 2.2.14
* @see #recordToString(ConsumerRecord)
*/
public static void setLogOnlyMetadata(boolean onlyMeta) {
LOG_METADATA_ONLY.set(onlyMeta);
}
/**
* Return the {@link ConsumerRecord} as a String; either {@code toString()} or
* {@code topic-partition@offset}.
* @param record the record.
* @return the rendered record.
* @since 2.2.14
* @see #setLogOnlyMetadata(boolean)
*/
public static String recordToString(ConsumerRecord<?, ?> record) {
if (Boolean.TRUE.equals(LOG_METADATA_ONLY.get())) {
return record.topic() + "-" + record.partition() + "@" + record.offset();
}
else {
return record.toString();
}
}
}

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

@@ -27,6 +27,8 @@ import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.listener.ListenerUtils;
/**
* Seek utilities.
*
@@ -68,11 +70,13 @@ public final class SeekUtils {
skipped.set(test);
}
catch (Exception ex) {
logger.error("Failed to determine if this record should be recovererd, including in seeks", ex);
logger.error("Failed to determine if this record ("
+ ListenerUtils.recordToString(record)
+ ") should be recovererd, including in seeks", ex);
skipped.set(false);
}
if (skipped.get() && logger.isDebugEnabled()) {
logger.debug("Skipping seek of: " + record);
logger.debug("Skipping seek of: " + ListenerUtils.recordToString(record));
}
}
if (!recoverable || !first.get() || !skipped.get()) {

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

@@ -48,6 +48,8 @@ See <<events>> for more information.
The `SeekToCurrentErrorHandler` can now be configured to commit the offset of a recovered record when the container is configured with `AckMode.MANUAL_IMMEDIATE` (since 2.2.4).
See <<seek-to-current>> for more information.
You can now suppress logging entire `ConsumerRecord` s in error, debug logs etc., by setting the `onlyLogRecordMetadata` container property to `true`.
==== @KafkaListener Changes
You can now override the `concurrency` and `autoStartup` properties of the listener container factory by setting properties on the annotation.