GH-1499: Option to suppress ConsumerRecord logging

Resolves #1499

**I will do the backports because I expect many conflicts.**

Author: garyrussell

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

@@ -222,6 +222,8 @@ public enum AckMode {
 
 	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 void setConsumerProperties(Properties consumerProperties) {
 		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 ["

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

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

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

@@ -692,6 +692,7 @@ public boolean isLongLived() {
 
 		@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 @@ private void handleAcks() {
 			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 @@ private void invokeRecordListenerInTx(final ConsumerRecords<K, V> records) {
 			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 @@ private void doInvokeWithRecords(final ConsumerRecords<K, V> records) {
 			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 @@ record = this.recordInterceptor.intercept(record);
 			}
 			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 {

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 @@ private 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 @@ else if (listener instanceof GenericMessageListener) {
 		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();
+		}
+	}
+
 }

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

@@ -27,6 +27,8 @@
 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 static boolean doSeeks(List<ConsumerRecord<?, ?>> records, Consumer<?, ?>
 					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()) {

src/reference/asciidoc/whats-new.adoc

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