Fix KafkaMessageSource for the latest SK

* Add `mockito-inline` dep to be able to mock `final` classes, e.g. `record`

Author: artembilan

build.gradle

@@ -84,7 +84,7 @@ ext {
     log4jVersion = '2.17.1'
     mailVersion = '2.0.1'
 	micrometerVersion = '2.0.0-SNAPSHOT'
-    mockitoVersion = '4.2.0'
+    mockitoVersion = '4.3.1'
     mongoDriverVersion = '4.5.0'
     mysqlVersion = '8.0.28'
     pahoMqttClientVersion = '1.2.5'
@@ -158,6 +158,7 @@ allprojects {
             mavenBom "io.projectreactor:reactor-bom:$reactorVersion"
             mavenBom "org.apache.logging.log4j:log4j-bom:$log4jVersion"
             mavenBom "org.springframework.data:spring-data-bom:$springDataVersion"
+            mavenBom "org.mockito:mockito-bom:$mockitoVersion"
         }
 
     }
@@ -445,7 +446,8 @@ project('spring-integration-test-support') {
     dependencies {
         compileOnly 'org.apiguardian:apiguardian-api:1.0.0'
         api "org.hamcrest:hamcrest-library:$hamcrestVersion"
-        api "org.mockito:mockito-core:$mockitoVersion"
+        api 'org.mockito:mockito-core'
+        api 'org.mockito:mockito-inline'
         api "org.assertj:assertj-core:$assertjVersion"
         api 'org.springframework:spring-context'
         api 'org.springframework:spring-messaging'

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.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.
@@ -55,10 +55,10 @@
 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;
+import org.springframework.kafka.support.KafkaUtils;
 import org.springframework.kafka.support.LogIfLevelEnabled;
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.kafka.support.converter.KafkaMessageHeaders;
@@ -656,21 +656,13 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 
 	/**
 	 * AcknowledgmentCallbackFactory for KafkaAckInfo.
+	 * @param consumerProperties the properties.
 	 * @param <K> the key type.
 	 * @param <V> the value type.
 	 *
 	 */
-	public static class KafkaAckCallbackFactory<K, V> implements AcknowledgmentCallbackFactory<KafkaAckInfo<K, V>> {
-
-		private final ConsumerProperties consumerProperties;
-
-		/**
-		 * Construct an instance with the provided properties.
-		 * @param consumerProperties the properties.
-		 */
-		public KafkaAckCallbackFactory(ConsumerProperties consumerProperties) {
-			this.consumerProperties = consumerProperties;
-		}
+	public record KafkaAckCallbackFactory<K, V>(ConsumerProperties consumerProperties)
+			implements AcknowledgmentCallbackFactory<KafkaAckInfo<K, V>> {
 
 		@Override
 		public AcknowledgmentCallback createCallback(KafkaAckInfo<K, V> info) {
@@ -737,15 +729,10 @@ public void acknowledge(Status status) {
 				try {
 					ConsumerRecord<K, V> record = this.ackInfo.getRecord();
 					switch (status) {
-						case ACCEPT:
-						case REJECT:
-							commitIfPossible(record);
-							break;
-						case REQUEUE:
-							rollback(record);
-							break;
-						default:
-							break;
+						case ACCEPT, REJECT -> commitIfPossible(record);
+						case REQUEUE -> rollback(record);
+						default -> {
+						}
 					}
 				}
 				catch (WakeupException e) {
@@ -774,7 +761,8 @@ private void rollback(ConsumerRecord<K, V> record) {
 									})
 									.collect(Collectors.toList());
 					if (rewound.size() > 0) {
-						this.logger.warn(() -> "Rolled back " + ListenerUtils.recordToString(record, this.logOnlyMetadata)
+						KafkaUtils.setLogOnlyMetadata(this.logOnlyMetadata);
+						this.logger.warn(() -> "Rolled back " + KafkaUtils.format(record)
 								+ " later in-flight offsets "
 								+ rewound + " will also be re-fetched");
 					}
@@ -783,9 +771,10 @@ private void rollback(ConsumerRecord<K, V> record) {
 		}
 
 		private void commitIfPossible(ConsumerRecord<K, V> record) { // NOSONAR
+			KafkaUtils.setLogOnlyMetadata(this.logOnlyMetadata);
 			if (this.ackInfo.isRolledBack()) {
 				this.logger.warn(() -> "Cannot commit offset for "
-						+ ListenerUtils.recordToString(record, this.logOnlyMetadata)
+						+ KafkaUtils.format(record)
 						+ "; an earlier offset was rolled back");
 			}
 			else {
@@ -809,16 +798,14 @@ private void commitIfPossible(ConsumerRecord<K, V> record) { // NOSONAR
 							ackInformation = toCommit.get(toCommit.size() - 1);
 							KafkaAckInfo<K, V> ackInformationToLog = ackInformation;
 							this.commitLogger.log(() -> "Committing pending offsets for "
-									+ ListenerUtils.recordToString(record, this.logOnlyMetadata)
+									+ KafkaUtils.format(record)
 									+ " and all deferred to "
-									+ ListenerUtils.recordToString(ackInformationToLog.getRecord(),
-									this.logOnlyMetadata));
-							candidates.removeAll(toCommit);
+									+ KafkaUtils.format(ackInformationToLog.getRecord()));
+							toCommit.forEach(candidates::remove);
 						}
 						else {
 							ackInformation = this.ackInfo;
-							this.commitLogger.log(() -> "Committing offset for "
-									+ ListenerUtils.recordToString(record, this.logOnlyMetadata));
+							this.commitLogger.log(() -> "Committing offset for " + KafkaUtils.format(record));
 						}
 					}
 					else { // earlier offsets present