diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java index 7912e4b4..3d605399 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java @@ -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 [" 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 9fc422cf..77c06f06 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 @@ -48,7 +48,8 @@ class FailedRecordTracker { FailedRecordTracker(@Nullable BiConsumer, 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; 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 7963169f..1006ae06 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 @@ -692,6 +692,7 @@ public class KafkaMessageListenerContainer // 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 // NOSONAR comment density ConsumerRecord 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 // NOSONAR comment density while (iterator.hasNext()) { final ConsumerRecord 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 // NOSONAR comment density while (iterator.hasNext()) { final ConsumerRecord 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 // 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 { 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 198ff2ef..663be16e 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 @@ -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 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(); + } + } + } 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 595132c4..57d866c1 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 @@ -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()) { diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 9a2cebff..e5f76f47 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -48,6 +48,8 @@ See <> 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 <> 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.