From 9223613ee19058f30a4fcc5c5910c4b0c54a5533 Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Wed, 8 Jul 2020 17:21:20 -0400
Subject: [PATCH] ConsumerRecord Logging Metadata Option

---
 build.gradle                                  |  2 +-
 .../kafka/inbound/KafkaMessageSource.java     | 20 ++++++++++++++-----
 2 files changed, 16 insertions(+), 6 deletions(-)

diff --git a/build.gradle b/build.gradle
index 0f07958d1b..d6a47a742e 100644
--- a/build.gradle
+++ b/build.gradle
@@ -98,7 +98,7 @@ ext {
 	soapVersion = '1.4.0'
 	springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.3.0-M1'
 	springDataVersion = project.hasProperty('springDataVersion') ? project.springDataVersion : '2020.0.0-M1'
-	springKafkaVersion = '2.5.3.RELEASE'
+	springKafkaVersion = '2.5.4.BUILD-SNAPSHOT'
 	springSecurityVersion = project.hasProperty('springSecurityVersion') ? project.springSecurityVersion : '5.4.0-M1'
 	springRetryVersion = '1.3.0'
 	springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.3.0-M1'
diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
index 09a4244887..f41e2646c3 100644
--- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
+++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
@@ -55,6 +55,7 @@ import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.listener.ConsumerAwareRebalanceListener;
 import org.springframework.kafka.listener.ConsumerProperties;
+import org.springframework.kafka.listener.ListenerUtils;
 import org.springframework.kafka.listener.LoggingCommitCallback;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.KafkaHeaders;
@@ -667,6 +668,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 
 		private final boolean isSyncCommits;
 
+		private final boolean logOnlyMetadata;
+
 		private volatile boolean acknowledged;
 
 		private boolean autoAckEnabled = true;
@@ -690,6 +693,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 					consumerProperties != null
 							? consumerProperties.getCommitLogLevel()
 							: LogIfLevelEnabled.Level.DEBUG);
+			this.logOnlyMetadata = consumerProperties.isOnlyLogRecordMetadata();
 		}
 
 		@Override
@@ -739,7 +743,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 									})
 									.collect(Collectors.toList());
 					if (rewound.size() > 0 && this.logger.isWarnEnabled()) {
-						this.logger.warn("Rolled back " + record + " later in-flight offsets "
+						this.logger.warn("Rolled back " + ListenerUtils.recordToString(record, this.logOnlyMetadata)
+								+ " later in-flight offsets "
 								+ rewound + " will also be re-fetched");
 					}
 				}
@@ -749,7 +754,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 		private void commitIfPossible(ConsumerRecord<K, V> record) {
 			if (this.ackInfo.isRolledBack()) {
 				if (this.logger.isWarnEnabled()) {
-					this.logger.warn("Cannot commit offset for " + record
+					this.logger.warn("Cannot commit offset for "
+							+ ListenerUtils.recordToString(record, this.logOnlyMetadata)
 							+ "; an earlier offset was rolled back");
 				}
 			}
@@ -773,13 +779,17 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 						if (toCommit.size() > 0) {
 							ackInformation = toCommit.get(toCommit.size() - 1);
 							KafkaAckInfo<K, V> ackInformationToLog = ackInformation;
-							this.commitLogger.log(() -> "Committing pending offsets for " + record
-									+ " and all deferred to " + ackInformationToLog.getRecord());
+							this.commitLogger.log(() -> "Committing pending offsets for "
+									+ ListenerUtils.recordToString(record, this.logOnlyMetadata)
+									+ " and all deferred to "
+									+ ListenerUtils.recordToString(ackInformationToLog.getRecord(),
+											this.logOnlyMetadata));
 							candidates.removeAll(toCommit);
 						}
 						else {
 							ackInformation = this.ackInfo;
-							this.commitLogger.log(() -> "Committing offset for " + record);
+							this.commitLogger.log(() -> "Committing offset for "
+									+ ListenerUtils.recordToString(record, this.logOnlyMetadata));
 						}
 					}
 					else { // earlier offsets present