GH-2153: Switch to ConsumerRecord Formatter

Resolves #2153

Author: garyrussell

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.
@@ -22,6 +22,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 
 import org.springframework.core.log.LogAccessor;
+import org.springframework.kafka.support.KafkaUtils;
 
 /**
  * The {@link CommonErrorHandler} implementation for logging exceptions.
@@ -50,7 +51,7 @@ public void setAckAfterHandle(boolean ackAfterHandle) {
 	public void handleRecord(Exception thrownException, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer,
 			MessageListenerContainer container) {
 
-		LOGGER.error(thrownException, () -> "Error occured while processing: " + ListenerUtils.recordToString(record));
+		LOGGER.error(thrownException, () -> "Error occured while processing: " + KafkaUtils.format(record));
 	}
 
 	@Override
@@ -59,7 +60,7 @@ public void handleBatch(Exception thrownException, ConsumerRecords<?, ?> data, C
 
 		StringBuilder message = new StringBuilder("Error occurred while processing:\n");
 		for (ConsumerRecord<?, ?> record : data) {
-			message.append(ListenerUtils.recordToString(record)).append('\n');
+			message.append(KafkaUtils.format(record)).append('\n');
 		}
 		LOGGER.error(thrownException, () -> message.substring(0, message.length() - 1));
 	}

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

@@ -332,14 +332,14 @@ public void accept(ConsumerRecord<?, ?> record, @Nullable Consumer<?, ?> consume
 		TopicPartition tp = this.destinationResolver.apply(record, exception);
 		if (tp == null) {
 			maybeThrow(record, exception);
-			this.logger.debug(() -> "Recovery of " + ListenerUtils.recordToString(record, true)
+			this.logger.debug(() -> "Recovery of " + KafkaUtils.format(record)
 					+ " skipped because destination resolver returned null");
 			return;
 		}
 		if (this.skipSameTopicFatalExceptions
 				&& tp.topic().equals(record.topic())
 				&& !getClassifier().classify(exception)) {
-			this.logger.error("Recovery of " + ListenerUtils.recordToString(record, true)
+			this.logger.error("Recovery of " + KafkaUtils.format(record)
 					+ " skipped because not retryable exception " + exception.toString()
 					+ " and the destination resolver routed back to the same topic");
 			return;
@@ -394,7 +394,7 @@ private void sendOrThrow(ProducerRecord<Object, Object> outRecord,
 
 	private void maybeThrow(ConsumerRecord<?, ?> record, Exception exception) {
 		String message = String.format("No destination returned for record %s and exception %s. " +
-				"failIfNoDestinationReturned: %s", ListenerUtils.recordToString(record), exception,
+				"failIfNoDestinationReturned: %s", KafkaUtils.format(record), exception,
 				this.throwIfNoDestinationReturned);
 		this.logger.warn(message);
 		if (this.throwIfNoDestinationReturned) {
@@ -513,7 +513,7 @@ protected void publish(ProducerRecord<Object, Object> outRecord, KafkaOperations
 			sendResult = kafkaTemplate.send(outRecord);
 			sendResult.addCallback(result -> {
 				this.logger.debug(() -> "Successful dead-letter publication: "
-						+ ListenerUtils.recordToString(inRecord, true) + " to " + result.getRecordMetadata());
+						+ KafkaUtils.format(inRecord) + " to " + result.getRecordMetadata());
 			}, ex -> {
 				this.logger.error(ex, () -> pubFailMessage(outRecord, inRecord));
 			});
@@ -548,7 +548,7 @@ private void verifySendResult(KafkaOperations<Object, Object> kafkaTemplate,
 
 	private String pubFailMessage(ProducerRecord<Object, Object> outRecord, ConsumerRecord<?, ?> inRecord) {
 		return "Dead-letter publication to "
-				+ outRecord.topic() + "failed for: " + ListenerUtils.recordToString(inRecord, true);
+				+ outRecord.topic() + "failed for: " + KafkaUtils.format(inRecord);
 	}
 
 	private Duration determineSendTimeout(KafkaOperations<?, ?> template) {

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.
@@ -24,6 +24,7 @@
 
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.KafkaException;
+import org.springframework.kafka.support.KafkaUtils;
 import org.springframework.util.backoff.BackOff;
 import org.springframework.util.backoff.BackOffExecution;
 
@@ -108,7 +109,7 @@ public static void retryBatch(Exception thrownException, ConsumerRecords<?, ?> r
 	public static String recordsToString(ConsumerRecords<?, ?> records) {
 		StringBuffer sb = new StringBuffer();
 		records.spliterator().forEachRemaining(rec -> sb
-				.append(ListenerUtils.recordToString(rec, true))
+				.append(KafkaUtils.format(rec))
 				.append(','));
 		sb.deleteCharAt(sb.length() - 1);
 		return sb.toString();

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

@@ -26,6 +26,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
 import org.springframework.core.log.LogAccessor;
+import org.springframework.kafka.support.KafkaUtils;
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.lang.Nullable;
 import org.springframework.util.backoff.BackOff;
@@ -152,7 +153,7 @@ protected RecoveryStrategy getRecoveryStrategy(List<ConsumerRecord<?, ?>> record
 			catch (Exception ex) {
 				if (records.size() > 0) {
 					this.logger.error(ex, () -> "Recovery of record ("
-							+ ListenerUtils.recordToString(records.get(0)) + ") failed");
+							+ KafkaUtils.format(records.get(0)) + ") failed");
 					this.failureTracker.getRetryListeners().forEach(rl ->
 							rl.recoveryFailed(records.get(0), thrownException, ex));
 				}

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

@@ -30,6 +30,7 @@
 import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.core.log.LogAccessor;
+import org.springframework.kafka.support.KafkaUtils;
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -76,7 +77,7 @@ class FailedRecordTracker implements RecoveryStrategy {
 					+ (failedRecord == null
 						? "none"
 						: failedRecord.getBackOffExecution())
-					+ " exhausted for " + ListenerUtils.recordToString(rec));
+					+ " exhausted for " + KafkaUtils.format(rec));
 			};
 		}
 		else {

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

@@ -1228,7 +1228,6 @@ public boolean isLongLived() {
 
 		@Override // NOSONAR complexity
 		public void run() {
-			ListenerUtils.setLogOnlyMetadata(this.containerProperties.isOnlyLogRecordMetadata());
 			publishConsumerStartingEvent();
 			this.consumerThread = Thread.currentThread();
 			setupSeeks();
@@ -1809,7 +1808,7 @@ record = this.acks.poll();
 		}
 
 		private void traceAck(ConsumerRecord<K, V> record) {
-			this.logger.trace(() -> "Ack: " + ListenerUtils.recordToString(record, true));
+			this.logger.trace(() -> "Ack: " + KafkaUtils.format(record));
 		}
 
 		private void doAck(ConsumerRecord<K, V> record) {
@@ -1905,14 +1904,14 @@ private synchronized void ackInOrder(ConsumerRecord<K, V> record) {
 				}
 				else if (record.offset() < offs.get(0)) {
 					throw new IllegalStateException("First remaining offset for this batch is " + offs.get(0)
-							+ "; you are acknowledging a stale record: " + ListenerUtils.recordToString(record));
+							+ "; you are acknowledging a stale record: " + KafkaUtils.format(record));
 				}
 				else {
 					deferred.add(record);
 				}
 			}
 			else {
-				throw new IllegalStateException("Unexpected ack for " + ListenerUtils.recordToString(record)
+				throw new IllegalStateException("Unexpected ack for " + KafkaUtils.format(record)
 						+ "; offsets list is empty");
 			}
 		}
@@ -2311,7 +2310,7 @@ private void invokeRecordListenerInTx(final ConsumerRecords<K, V> records) {
 				if (record == null) {
 					continue;
 				}
-				this.logger.trace(() -> "Processing " + ListenerUtils.recordToString(record));
+				this.logger.trace(() -> "Processing " + KafkaUtils.format(record));
 				try {
 					invokeInTransaction(iterator, record);
 				}
@@ -2412,7 +2411,7 @@ private void doInvokeWithRecords(final ConsumerRecords<K, V> records) {
 				if (record == null) {
 					continue;
 				}
-				this.logger.trace(() -> "Processing " + ListenerUtils.recordToString(record));
+				this.logger.trace(() -> "Processing " + KafkaUtils.format(record));
 				doInvokeRecordListener(record, iterator);
 				if (this.commonRecordInterceptor !=  null) {
 					this.commonRecordInterceptor.afterRecord(record, this.consumer);
@@ -2445,7 +2444,7 @@ private ConsumerRecord<K, V> checkEarlyIntercept(ConsumerRecord<K, V> recordArg)
 				record = this.earlyRecordInterceptor.intercept(record, this.consumer);
 				if (record == null) {
 					this.logger.debug(() -> "RecordInterceptor returned null, skipping: "
-						+ ListenerUtils.recordToString(recordArg));
+						+ KafkaUtils.format(recordArg));
 				}
 			}
 			return record;
@@ -2604,7 +2603,7 @@ record = this.recordInterceptor.intercept(record, this.consumer);
 			}
 			if (record == null) {
 				this.logger.debug(() -> "RecordInterceptor returned null, skipping: "
-						+ ListenerUtils.recordToString(recordArg));
+						+ KafkaUtils.format(recordArg));
 			}
 			else {
 				try {
@@ -3158,7 +3157,7 @@ public void nack(long sleep) {
 
 			@Override
 			public String toString() {
-				return "Acknowledgment for " + ListenerUtils.recordToString(this.record, true);
+				return "Acknowledgment for " + KafkaUtils.format(this.record);
 			}
 
 		}

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

@@ -27,7 +27,6 @@
 import org.apache.kafka.common.header.internals.RecordHeaders;
 
 import org.springframework.core.log.LogAccessor;
-import org.springframework.kafka.support.KafkaUtils;
 import org.springframework.kafka.support.serializer.DeserializationException;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -143,40 +142,6 @@ 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
-	 * @see #recordToString(ConsumerRecord)
-	 */
-	public static void setLogOnlyMetadata(boolean onlyMeta) {
-		KafkaUtils.setLogOnlyMetadata(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) {
-		return KafkaUtils.recordToString(record);
-	}
-
-	/**
-	 * Return the {@link ConsumerRecord} as a String; either {@code toString()} or
-	 * {@code topic-partition@offset}.
-	 * @param record the record.
-	 * @param meta true to log just the metadata.
-	 * @return the rendered record.
-	 * @since 2.5.4
-	 */
-	public static String recordToString(ConsumerRecord<?, ?> record, boolean meta) {
-		return KafkaUtils.recordToString(record, meta);
-	}
-
 	/**
 	 * Sleep according to the {@link BackOff}; when the {@link BackOffExecution} returns
 	 * {@link BackOffExecution#STOP} sleep for the previous backOff.

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.
@@ -35,6 +35,7 @@
 import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.KafkaException.Level;
 import org.springframework.kafka.listener.ContainerProperties.AckMode;
+import org.springframework.kafka.support.KafkaUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.backoff.FixedBackOff;
@@ -107,18 +108,18 @@ public static boolean doSeeks(List<ConsumerRecord<?, ?>> records, Consumer<?, ?>
 				}
 				catch (Exception ex) {
 					if (isBackoffException(ex)) {
-						logger.debug(ex, () -> ListenerUtils.recordToString(record)
+						logger.debug(ex, () -> KafkaUtils.format(record)
 								+ " included in seeks due to retry back off");
 					}
 					else {
 						logger.error(ex, () -> "Failed to determine if this record ("
-								+ ListenerUtils.recordToString(record)
+								+ KafkaUtils.format(record)
 								+ ") should be recovererd, including in seeks");
 					}
 					skipped.set(false);
 				}
 				if (skipped.get()) {
-					logger.debug(() -> "Skipping seek of: " + ListenerUtils.recordToString(record));
+					logger.debug(() -> "Skipping seek of: " + KafkaUtils.format(record));
 				}
 			}
 			if (!recoverable || !first.get() || !skipped.get()) {

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";
 	}