GH-2132: Deprecate ListenerUtils.recordToString

Resolves #2133

Apply to 2.8.x, and 2.7.x.

In 3.0 they will be removed and `KafkaUtils` called directly instead.

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -2560,6 +2560,8 @@ See `monitorInterval`.
 |[[onlyLogRecordMetadata]]<<onlyLogRecordMetadata,`onlyLogRecordMetadata`>>
 |`false`
 |Set to false to log the complete consumer record (in error, debug logs etc) instead of just `topic-partition@offset`.
+Deprecated.
+Replaced by `KafkaUtils.setConsumerRecordFormatter`.
 
 |[[pollTimeout]]<<pollTimeout,`pollTimeout`>>
 |5000
@@ -5720,3 +5722,10 @@ public KafkaJaasLoginModuleInitializer jaasConfig() throws IOException {
 }
 ----
 ====
+
+[[record-logging]]
+==== Producer and Consumer Record Logging
+
+Starting with versions 2.7.12, 2.8.4, you can determine how these records will be rendered in debug logs, etc.
+
+See `KafkaUtils.setProducerRecordFormatter()` and `KafkaUtils.setProducerRecordFormatter()` for more information.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -46,13 +46,15 @@ public void setAckAfterHandle(boolean ackAfterHandle) {
 		this.ackAfterHandle = ackAfterHandle;
 	}
 
+	@SuppressWarnings("deprecation")
 	@Override
 	public void handleRecord(Exception thrownException, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer,
 			MessageListenerContainer container) {
 
 		LOGGER.error(thrownException, () -> "Error occured while processing: " + ListenerUtils.recordToString(record));
 	}
 
+	@SuppressWarnings("deprecation")
 	@Override
 	public void handleBatch(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
 			MessageListenerContainer container, Runnable invokeListener) {

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

@@ -24,6 +24,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.OffsetCommitCallback;
 
+import org.springframework.kafka.support.KafkaUtils;
 import org.springframework.kafka.support.LogIfLevelEnabled;
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.lang.Nullable;
@@ -431,16 +432,20 @@ public void setCommitRetries(int commitRetries) {
 		this.commitRetries = commitRetries;
 	}
 
+	@Deprecated
 	public boolean isOnlyLogRecordMetadata() {
 		return this.onlyLogRecordMetadata;
 	}
 
 	/**
-	 * Set to false to log {@code record.toString()} in log messages instead
-	 * of {@code topic-partition@offset}.
+	 * Set to false to log {@code record.toString()} in log messages instead of
+	 * {@code topic-partition@offset}.
 	 * @param onlyLogRecordMetadata false to log the entire record.
 	 * @since 2.2.14
+	 * @deprecated in favor of
+	 * {@link KafkaUtils#setConsumerRecordFormatter(java.util.function.Function)}.
 	 */
+	@Deprecated
 	public void setOnlyLogRecordMetadata(boolean onlyLogRecordMetadata) {
 		this.onlyLogRecordMetadata = onlyLogRecordMetadata;
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -339,7 +339,7 @@ public void setSkipSameTopicFatalExceptions(boolean skipSameTopicFatalExceptions
 		this.skipSameTopicFatalExceptions = skipSameTopicFatalExceptions;
 	}
 
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings({ "unchecked", "deprecation" })
 	@Override
 	public void accept(ConsumerRecord<?, ?> record, @Nullable Consumer<?, ?> consumer, Exception exception) {
 		TopicPartition tp = this.destinationResolver.apply(record, exception);
@@ -406,6 +406,7 @@ private void sendOrThrow(ProducerRecord<Object, Object> outRecord,
 	}
 
 	private void maybeThrow(ConsumerRecord<?, ?> record, Exception exception) {
+		@SuppressWarnings("deprecation")
 		String message = String.format("No destination returned for record %s and exception %s. " +
 				"failIfNoDestinationReturned: %s", ListenerUtils.recordToString(record), exception,
 				this.throwIfNoDestinationReturned);
@@ -518,6 +519,7 @@ protected ProducerRecord<Object, Object> createProducerRecord(ConsumerRecord<?,
 	 * @param inRecord the consumer record.
 	 * @since 2.2.5
 	 */
+	@SuppressWarnings("deprecation")
 	protected void publish(ProducerRecord<Object, Object> outRecord, KafkaOperations<Object, Object> kafkaTemplate,
 			ConsumerRecord<?, ?> inRecord) {
 
@@ -559,6 +561,7 @@ private void verifySendResult(KafkaOperations<Object, Object> kafkaTemplate,
 		}
 	}
 
+	@SuppressWarnings("deprecation")
 	private String pubFailMessage(ProducerRecord<Object, Object> outRecord, ConsumerRecord<?, ?> inRecord) {
 		return "Dead-letter publication to "
 				+ outRecord.topic() + "failed for: " + ListenerUtils.recordToString(inRecord, true);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -105,6 +105,7 @@ public static void retryBatch(Exception thrownException, ConsumerRecords<?, ?> r
 	 * @param records the records.
 	 * @return the String.
 	 */
+	@SuppressWarnings("deprecation")
 	public static String recordsToString(ConsumerRecords<?, ?> records) {
 		StringBuffer sb = new StringBuffer();
 		records.spliterator().forEachRemaining(rec -> sb

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2021 the original author or authors.
+ * Copyright 2019-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -126,6 +126,7 @@ public int deliveryAttempt(TopicPartitionOffset topicPartitionOffset) {
 	 * @return the {@link BiPredicate}.
 	 * @deprecated in favor of {@link #getRecoveryStrategy(List, Exception)}.
 	 */
+	@SuppressWarnings("deprecation")
 	@Deprecated
 	protected BiPredicate<ConsumerRecord<?, ?>, Exception> getSkipPredicate(List<ConsumerRecord<?, ?>> records,
 			Exception thrownException) {
@@ -169,6 +170,7 @@ protected RecoveryStrategy getRecoveryStrategy(List<ConsumerRecord<?, ?>> record
 	 * @return the {@link RecoveryStrategy}.
 	 * @since 2.8.4
 	 */
+	@SuppressWarnings("deprecation")
 	protected RecoveryStrategy getRecoveryStrategy(List<ConsumerRecord<?, ?>> records,
 												@Nullable Consumer<?, ?> recoveryConsumer, Exception thrownException) {
 		if (getClassifier().classify(thrownException)) {

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

@@ -61,6 +61,7 @@ class FailedRecordTracker implements RecoveryStrategy {
 
 	private boolean resetStateOnExceptionChange;
 
+	@SuppressWarnings("deprecation")
 	FailedRecordTracker(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff,
 			LogAccessor logger) {
 

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

@@ -1229,6 +1229,7 @@ public boolean isLongLived() {
 			return true;
 		}
 
+		@SuppressWarnings("deprecation")
 		@Override // NOSONAR complexity
 		public void run() {
 			ListenerUtils.setLogOnlyMetadata(this.containerProperties.isOnlyLogRecordMetadata());
@@ -1811,6 +1812,7 @@ record = this.acks.poll();
 			}
 		}
 
+		@SuppressWarnings("deprecation")
 		private void traceAck(ConsumerRecord<K, V> record) {
 			this.logger.trace(() -> "Ack: " + ListenerUtils.recordToString(record, true));
 		}
@@ -1885,6 +1887,7 @@ private void processAcks(ConsumerRecords<K, V> records) {
 			}
 		}
 
+		@SuppressWarnings("deprecation")
 		private synchronized void ackInOrder(ConsumerRecord<K, V> record) {
 			TopicPartition part = new TopicPartition(record.topic(), record.partition());
 			List<Long> offs = this.offsetsInThisBatch.get(part);
@@ -2303,7 +2306,7 @@ private void invokeRecordListener(final ConsumerRecords<K, V> records) {
 		 * Invoke the listener with each record in a separate transaction.
 		 * @param records the records.
 		 */
-		@SuppressWarnings(RAW_TYPES) // NOSONAR complexity
+		@SuppressWarnings("deprecation") // NOSONAR complexity
 		private void invokeRecordListenerInTx(final ConsumerRecords<K, V> records) {
 			Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
 			while (iterator.hasNext()) {
@@ -2405,6 +2408,7 @@ protected void doInTransactionWithoutResult(TransactionStatus status) {
 			}
 		}
 
+		@SuppressWarnings("deprecation")
 		private void doInvokeWithRecords(final ConsumerRecords<K, V> records) {
 			Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
 			while (iterator.hasNext()) {
@@ -2440,6 +2444,7 @@ private ConsumerRecords<K, V> checkEarlyIntercept(ConsumerRecords<K, V> nextArg)
 			return next;
 		}
 
+		@SuppressWarnings("deprecation")
 		@Nullable
 		private ConsumerRecord<K, V> checkEarlyIntercept(ConsumerRecord<K, V> recordArg) {
 			internalHeaders(recordArg);
@@ -2600,14 +2605,15 @@ private void invokeOnMessage(final ConsumerRecord<K, V> record) {
 			}
 		}
 
+		@SuppressWarnings("deprecation")
 		private void doInvokeOnMessage(final ConsumerRecord<K, V> recordArg) {
 			ConsumerRecord<K, V> record = recordArg;
 			if (this.recordInterceptor != null) {
 				record = this.recordInterceptor.intercept(record, this.consumer);
 			}
 			if (record == null) {
-				this.logger.debug(() -> "RecordInterceptor returned null, skipping: "
-						+ ListenerUtils.recordToString(recordArg));
+				this.logger.debug(() -> ("RecordInterceptor returned null, skipping: "
+						+ ListenerUtils.recordToString(recordArg)));
 			}
 			else {
 				try {
@@ -3165,6 +3171,7 @@ public void nack(long sleep) {
 			}
 
 			@Override
+			@SuppressWarnings("deprecation")
 			public String toString() {
 				return "Acknowledgment for " + ListenerUtils.recordToString(this.record, true);
 			}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2021 the original author or authors.
+ * Copyright 2017-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -49,6 +49,8 @@ public final class ListenerUtils {
 	private ListenerUtils() {
 	}
 
+	private static final ThreadLocal<Boolean> LOG_METADATA_ONLY = new ThreadLocal<>();
+
 	private static final int DEFAULT_SLEEP_INTERVAL = 100;
 
 	private static final int SMALL_SLEEP_INTERVAL = 10;
@@ -150,10 +152,12 @@ protected Class<?> resolveClass(ObjectStreamClass desc) throws IOException, Clas
 	 * Set to true to only log record metadata.
 	 * @param onlyMeta true to only log record metadata.
 	 * @since 2.2.14
+	 * @deprecated in favor of {@link KafkaUtils#format(ConsumerRecord)}.
 	 * @see #recordToString(ConsumerRecord)
 	 */
+	@Deprecated
 	public static void setLogOnlyMetadata(boolean onlyMeta) {
-		KafkaUtils.setLogOnlyMetadata(onlyMeta);
+		LOG_METADATA_ONLY.set(onlyMeta);
 	}
 
 	/**
@@ -162,10 +166,12 @@ public static void setLogOnlyMetadata(boolean onlyMeta) {
 	 * @param record the record.
 	 * @return the rendered record.
 	 * @since 2.2.14
+	 * @deprecated in favor of {@link KafkaUtils#format(ConsumerRecord)}.
 	 * @see #setLogOnlyMetadata(boolean)
 	 */
+	@Deprecated
 	public static String recordToString(ConsumerRecord<?, ?> record) {
-		return KafkaUtils.recordToString(record);
+		return recordToString(record, Boolean.TRUE.equals(LOG_METADATA_ONLY.get()));
 	}
 
 	/**
@@ -175,9 +181,11 @@ public static String recordToString(ConsumerRecord<?, ?> record) {
 	 * @param meta true to log just the metadata.
 	 * @return the rendered record.
 	 * @since 2.5.4
+	 * @deprecated in favor of {@link KafkaUtils#format(ConsumerRecord)}.
 	 */
+	@Deprecated
 	public static String recordToString(ConsumerRecord<?, ?> record, boolean meta) {
-		return KafkaUtils.recordToString(record, meta);
+		return KafkaUtils.format(record, !meta);
 	}
 
 	/**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -92,6 +92,7 @@ public static boolean doSeeks(List<ConsumerRecord<?, ?>> records, Consumer<?, ?>
 	 * @param logger a {@link LogAccessor} for seek errors.
 	 * @return true if the failed record was skipped.
 	 */
+	@SuppressWarnings("deprecation")
 	public static boolean doSeeks(List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer, Exception exception,
 			boolean recoverable, RecoveryStrategy recovery, @Nullable MessageListenerContainer container,
 			LogAccessor logger) {

spring-kafka/src/main/java/org/springframework/kafka/requestreply/AggregatingReplyingKafkaTemplate.java

@@ -127,7 +127,7 @@ public void onMessage(List<ConsumerRecord<K, Collection<ConsumerRecord<K, R>>>>
 		data.forEach(record -> {
 			Header correlation = record.headers().lastHeader(KafkaHeaders.CORRELATION_ID);
 			if (correlation == null) {
-				this.logger.error(() -> "No correlationId found in reply: " + KafkaUtils.recordToString(record)
+				this.logger.error(() -> "No correlationId found in reply: " + KafkaUtils.format(record)
 						+ " - to use request/reply semantics, the responding server must return the correlation id "
 						+ " in the '" + KafkaHeaders.CORRELATION_ID + "' header");
 			}

spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java

@@ -457,7 +457,7 @@ public void onMessage(List<ConsumerRecord<K, R>> data) {
 				correlationId = new CorrelationKey(correlationHeader.value());
 			}
 			if (correlationId == null) {
-				this.logger.error(() -> "No correlationId found in reply: " + KafkaUtils.recordToString(record)
+				this.logger.error(() -> "No correlationId found in reply: " + KafkaUtils.format(record)
 						+ " - to use request/reply semantics, the responding server must return the correlation id "
 						+ " in the '" + this.correlationHeaderName + "' header");
 			}
@@ -475,7 +475,7 @@ public void onMessage(List<ConsumerRecord<K, R>> data) {
 						future.setException(exception);
 					}
 					if (ok) {
-						this.logger.debug(() -> "Received: " + KafkaUtils.recordToString(record)
+						this.logger.debug(() -> "Received: " + KafkaUtils.format(record)
 								+ WITH_CORRELATION_ID + correlationKey);
 						future.set(record);
 					}
@@ -543,7 +543,7 @@ protected void logLateArrival(ConsumerRecord<K, R> record, CorrelationKey correl
 	}
 
 	private String missingCorrelationLogMessage(ConsumerRecord<K, R> record, CorrelationKey correlationId) {
-		return "No pending reply: " + KafkaUtils.recordToString(record) + WITH_CORRELATION_ID
+		return "No pending reply: " + KafkaUtils.format(record) + WITH_CORRELATION_ID
 				+ correlationId + ", perhaps timed out, or using a shared reply topic";
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurer.java

@@ -438,6 +438,7 @@ static class LoggingDltListenerHandlerMethod {
 
 		public static final String DEFAULT_DLT_METHOD_NAME = "logMessage";
 
+		@SuppressWarnings("deprecation")
 		public void logMessage(Object message) {
 			if (message instanceof ConsumerRecord) {
 				LOGGER.info(() -> "Received message in dlt listener: "