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;
@@ -374,6 +376,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

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

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

@@ -63,7 +63,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

@@ -958,6 +958,7 @@ public boolean isLongLived() {
 
 		@Override
 		public void run() {
+			ListenerUtils.setLogOnlyMetadata(this.containerProperties.isOnlyLogRecordMetadata());
 			publishConsumerStartingEvent();
 			this.consumerThread = Thread.currentThread();
 			setupSeeks();
@@ -1280,7 +1281,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) {
@@ -1629,7 +1630,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 {
 					if (this.producerPerConsumerPartition) {
 						TransactionSupport
@@ -1714,7 +1715,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, iterator);
 				if (this.nackSleep >= 0) {
 					handleNack(records, record);
@@ -1728,7 +1729,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;
@@ -1843,7 +1845,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

@@ -42,6 +42,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;
@@ -96,4 +98,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/listener/SeekUtils.java

@@ -85,12 +85,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

@@ -655,6 +655,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

@@ -48,6 +48,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/kafka.adoc

@@ -2241,6 +2241,11 @@ See `noPollThreshold` and `pollTimeout`.
 |Multiplied by `pollTimeOut` to determine whether to publish a `NonResponsiveConsumerEvent`.
 See `monitorInterval`.
 
+|onlyLogRecord
+Metadata
+|`false`
+|Set to true to show only the `topic-partition@offset` for a record instead of the whole consumer record (in error, debug logs etc).
+
 |pollTimeout
 |5000
 |The timeout passed into `Consumer.poll()`.

src/reference/asciidoc/whats-new.adoc

@@ -67,6 +67,9 @@ See <<error-handlers>> for more information.
 The `getAssignmentsByClientId()` method has been added, making it easier to determine which consumers in a concurrent container are assigned which partition(s).
 See <<container-props>> for more information.
 
+You can now suppress logging entire `ConsumerRecord` s in error, debug logs etc.
+See `onlyLogRecordMetadata` in <<container-props>>.
+
 [[x25-template]]
 ==== KafkaTemplate Changes