Partial Revert "GH-2132: Producer Record Log Fo..."

Resolves #2132

This reverts commit 297f52d.

Just the consumer record part.

Author: garyrussell

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

@@ -29,7 +29,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;
@@ -48,6 +47,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;
@@ -120,7 +121,7 @@ public static DeserializationException getExceptionFromHeader(final ConsumerReco
 	 * @see #recordToString(ConsumerRecord)
 	 */
 	public static void setLogOnlyMetadata(boolean onlyMeta) {
-		KafkaUtils.setLogOnlyMetadata(onlyMeta);
+		LOG_METADATA_ONLY.set(onlyMeta);
 	}
 
 	/**
@@ -132,7 +133,12 @@ public static void setLogOnlyMetadata(boolean onlyMeta) {
 	 * @see #setLogOnlyMetadata(boolean)
 	 */
 	public static String recordToString(ConsumerRecord<?, ?> record) {
-		return KafkaUtils.recordToString(record);
+		if (Boolean.TRUE.equals(LOG_METADATA_ONLY.get())) {
+			return record.topic() + "-" + record.partition() + "@" + record.offset();
+		}
+		else {
+			return record.toString();
+		}
 	}
 
 	/**
@@ -144,7 +150,12 @@ public static String recordToString(ConsumerRecord<?, ?> record) {
 	 * @since 2.5.4
 	 */
 	public static String recordToString(ConsumerRecord<?, ?> record, boolean meta) {
-		return KafkaUtils.recordToString(record, meta);
+		if (meta) {
+			return record.topic() + "-" + record.partition() + "@" + record.offset();
+		}
+		else {
+			return record.toString();
+		}
 	}
 
 	/**

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/support/KafkaUtils.java

@@ -42,8 +42,6 @@
  */
 public final class KafkaUtils {
 
-	private static final ThreadLocal<Boolean> LOG_METADATA_ONLY = new ThreadLocal<>();
-
 	private static Function<ProducerRecord<?, ?>, String> prFormatter = rec -> rec.toString();
 
 	private static Function<ConsumerRecord<?, ?>, String> crFormatter =
@@ -145,49 +143,10 @@ else if (dt instanceof String) {
 				min));
 	}
 
-	/**
-	 * Set to true to only log record metadata.
-	 * @param onlyMeta true to only log record metadata.
-	 * @since 2.7.12
-	 * @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.7.12
-	 * @see #setLogOnlyMetadata(boolean)
-	 */
-	public static String recordToString(ConsumerRecord<?, ?> record) {
-		return recordToString(record, Boolean.TRUE.equals(LOG_METADATA_ONLY.get()));
-	}
-
-	/**
-	 * 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.7.12
-	 */
-	public static String recordToString(ConsumerRecord<?, ?> record, boolean meta) {
-		if (meta) {
-			return crFormatter.apply(record);
-		}
-		else {
-			return record.toString();
-		}
-	}
-
 	/**
 	 * Set a formatter for logging {@link ConsumerRecord}s.
 	 * @param formatter a function to format the record as a String
-	 * @since 2.7.11
+	 * @since 2.7.12
 	 */
 	public static void setConsumerRecordFormatter(Function<ConsumerRecord<?, ?>, String> formatter) {
 		Assert.notNull(formatter, "'formatter' cannot be null");
@@ -197,7 +156,7 @@ public static void setConsumerRecordFormatter(Function<ConsumerRecord<?, ?>, Str
 	/**
 	 * Set a formatter for logging {@link ProducerRecord}s.
 	 * @param formatter a function to format the record as a String
-	 * @since 2.7.11
+	 * @since 2.7.12
 	 */
 	public static void setProducerRecordFormatter(Function<ProducerRecord<?, ?>, String> formatter) {
 		Assert.notNull(formatter, "'formatter' cannot be null");
@@ -209,6 +168,7 @@ public static void setProducerRecordFormatter(Function<ProducerRecord<?, ?>, Str
 	 * {@code topic-partition@offset}.
 	 * @param record the record to format.
 	 * @return the formatted String.
+	 * @since 2.7.12
 	 */
 	public static String format(ConsumerRecord<?, ?> record) {
 		return crFormatter.apply(record);
@@ -219,6 +179,7 @@ public static String format(ConsumerRecord<?, ?> record) {
 	 * {@link ProducerRecord}{@link #toString()}.
 	 * @param record the record to format.
 	 * @return the formatted String.
+	 * @since 2.7.12
 	 */
 	public static String format(ProducerRecord<?, ?> record) {
 		return prFormatter.apply(record);