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/ConsumerProperties.java

@@ -97,6 +97,8 @@ public class ConsumerProperties {
 
 	private LogIfLevelEnabled.Level commitLogLevel = LogIfLevelEnabled.Level.DEBUG;
 
+	private boolean onlyLogRecordMetadata;
+
 	private Properties kafkaConsumerProperties = new Properties();
 
 	private Duration authorizationExceptionRetryInterval;
@@ -350,6 +352,20 @@ public void setCommitRetries(int commitRetries) {
 		this.commitRetries = commitRetries;
 	}
 
+	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 "ConsumerProperties ["

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

@@ -183,7 +183,10 @@ public boolean removeNotRetryableException(Class<? extends Exception> exceptionT
 				this.failureTracker.getRecoverer().accept(records.get(0), thrownException);
 			}
 			catch (Exception ex) {
-				this.logger.error(ex, () -> "Recovery of record (" + records.get(0) + ") failed");
+				if (records.size() > 0) {
+					this.logger.error(ex, () -> "Recovery of record ("
+							+ ListenerUtils.recordToString(records.get(0)) + ") failed");
+				}
 				return NEVER_SKIP_PREDICATE;
 			}
 			return ALWAYS_SKIP_PREDICATE;

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

@@ -61,7 +61,7 @@ class FailedRecordTracker {
 					+ (failedRecord == null
 						? "none"
 						: failedRecord.getBackOffExecution())
-					+ " exhausted for " + rec);
+					+ " exhausted for " + ListenerUtils.recordToString(rec));
 			};
 		}
 		else {

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

@@ -944,6 +944,7 @@ public boolean isLongLived() {
 
 		@Override
 		public void run() {
+			ListenerUtils.setLogOnlyMetadata(this.containerProperties.isOnlyLogRecordMetadata());
 			publishConsumerStartingEvent();
 			this.consumerThread = Thread.currentThread();
 			if (this.consumerSeekAwareListener != null) {
@@ -1248,7 +1249,7 @@ record = this.acks.poll();
 		}
 
 		private void traceAck(ConsumerRecord<K, V> record) {
-			this.logger.trace(() -> "Ack: " + record);
+			this.logger.trace(() -> "Ack: " + ListenerUtils.recordToString(record));
 		}
 
 		private void processAck(ConsumerRecord<K, V> record) {
@@ -1589,7 +1590,7 @@ private void invokeRecordListenerInTx(final ConsumerRecords<K, V> records) {
 				if (record == null) {
 					continue;
 				}
-				this.logger.trace(() -> "Processing " + record);
+				this.logger.trace(() -> "Processing " + ListenerUtils.recordToString(record));
 				try {
 					TransactionSupport
 							.setTransactionIdSuffix(zombieFenceTxIdSuffix(record.topic(), record.partition()));
@@ -1669,7 +1670,7 @@ private void doInvokeWithRecords(final ConsumerRecords<K, V> records) {
 				if (record == null) {
 					continue;
 				}
-				this.logger.trace(() -> "Processing " + record);
+				this.logger.trace(() -> "Processing " + ListenerUtils.recordToString(record));
 				doInvokeRecordListener(record, null, iterator);
 				if (this.nackSleep >= 0) {
 					handleNack(records, record);
@@ -1683,7 +1684,8 @@ private ConsumerRecord<K, V> checkEarlyIntercept(ConsumerRecord<K, V> nextArg) {
 			if (this.earlyRecordInterceptor != null) {
 				next = this.earlyRecordInterceptor.intercept(next);
 				if (next == null && this.logger.isDebugEnabled()) {
-					this.logger.debug("RecordInterceptor returned null, skipping: " + nextArg);
+					this.logger.debug("RecordInterceptor returned null, skipping: "
+						+ ListenerUtils.recordToString(nextArg));
 				}
 			}
 			return next;
@@ -1788,7 +1790,8 @@ private void doInvokeOnMessage(final ConsumerRecord<K, V> recordArg) {
 				record = this.recordInterceptor.intercept(record);
 			}
 			if (record == null) {
-				this.logger.debug(() -> "RecordInterceptor returned null, skipping: " + recordArg);
+				this.logger.debug(() -> "RecordInterceptor returned null, skipping: "
+						+ ListenerUtils.recordToString(recordArg));
 			}
 			else {
 				switch (this.listenerType) {

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

@@ -45,6 +45,8 @@ public final class ListenerUtils {
 	private ListenerUtils() {
 	}
 
+	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;
@@ -101,4 +103,32 @@ public static DeserializationException getExceptionFromHeader(final ConsumerReco
 		return null;
 	}
 
+	/**
+	 * 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,7 @@
 import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.core.log.LogAccessor;
+import org.springframework.kafka.listener.ListenerUtils;
 import org.springframework.util.backoff.FixedBackOff;
 
 /**
@@ -75,12 +76,12 @@ public static boolean doSeeks(List<ConsumerRecord<?, ?>> records, Consumer<?, ?>
 					skipped.set(test);
 				}
 				catch (Exception ex) {
-					logger.error(ex, "Failed to determine if this record (" + record
+					logger.error(ex, "Failed to determine if this record (" + ListenerUtils.recordToString(record)
 							+ ") should be recovererd, including in seeks");
 					skipped.set(false);
 				}
 				if (skipped.get()) {
-					logger.debug(() -> "Skipping seek of: " + record);
+					logger.debug(() -> "Skipping seek of: " + ListenerUtils.recordToString(record));
 				}
 			}
 			if (!recoverable || !first.get() || !skipped.get()) {

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

@@ -657,6 +657,7 @@ public void testRecordAckAfterRecoveryMock() throws Exception {
 		containerProps.setGroupId("grp");
 		containerProps.setAckMode(AckMode.RECORD);
 		containerProps.setMissingTopicsFatal(false);
+		containerProps.setOnlyLogRecordMetadata(true);
 		final CountDownLatch latch = new CountDownLatch(2);
 		MessageListener<Integer, String> messageListener = spy(
 				new MessageListener<Integer, String>() { // Cannot be lambda: Mockito doesn't mock final classes

spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentErrorHandlerTests.java

@@ -49,6 +49,7 @@ public class SeekToCurrentErrorHandlerTests {
 
 	@Test
 	public void testClassifier() {
+		ListenerUtils.setLogOnlyMetadata(true);
 		AtomicReference<ConsumerRecord<?, ?>> recovered = new AtomicReference<>();
 		AtomicBoolean recovererShouldFail = new AtomicBoolean(false);
 		SeekToCurrentErrorHandler handler = new SeekToCurrentErrorHandler((r, t) -> {

src/reference/asciidoc/whats-new.adoc

@@ -47,6 +47,8 @@ See its JavaDocs and <<kafka-container>> for more information.
 Static group membership is now supported.
 See <<message-listener-container>> 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
 
 The `@KafkaListener` annotation has a new property `splitIterables`; default true.