GH-2208: Fix Manual Nack with Mutating Interceptor

Resolves #2208

`equals()` test on `ConsumerRecord` fails (not implemented).

Compare topic, partition and offset instead.

**cherry-pick to 2.9.x, 2.8.x, 2.7.x**

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

Author: garyrussell

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

@@ -2201,14 +2201,21 @@ private void handleNack(final ConsumerRecords<K, V> records, final ConsumerRecor
 			Iterator<ConsumerRecord<K, V>> iterator2 = records.iterator();
 			while (iterator2.hasNext()) {
 				ConsumerRecord<K, V> next = iterator2.next();
-				if (next.equals(record) || list.size() > 0) {
+				if (list.size() > 0 || recordsEqual(record, next)) {
 					list.add(next);
 				}
 			}
 			SeekUtils.doSeeks(list, this.consumer, null, true, (rec, ex) -> false, this.logger); // NOSONAR
 			nackSleepAndReset();
 		}
 
+
+		private boolean recordsEqual(ConsumerRecord<K, V> rec1, ConsumerRecord<K, V> rec2) {
+			return rec1.topic().equals(rec2.topic())
+					&& rec1.partition() == rec2.partition()
+					&& rec1.offset() == rec2.offset();
+		}
+
 		private void nackSleepAndReset() {
 			try {
 				Thread.sleep(this.nackSleep);

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

@@ -37,7 +37,9 @@ public interface RecordInterceptor<K, V> {
 
 	/**
 	 * Perform some action on the record or return a different one. If null is returned
-	 * the record will be skipped. Invoked before the listener.
+	 * the record will be skipped. Invoked before the listener. IMPORTANT; if this method
+	 * returns a different record, the topic, partition and offset must not be changed
+	 * to avoid undesirable side-effects.
 	 * @param record the record.
 	 * @return the record or null.
 	 * @deprecated in favor of {@link #intercept(ConsumerRecord, Consumer)} which will

spring-kafka/src/test/java/org/springframework/kafka/listener/ManualNackRecordTests.java

@@ -58,6 +58,7 @@
 import org.springframework.kafka.listener.ContainerProperties.AckMode;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.lang.Nullable;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
@@ -212,6 +213,17 @@ public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
 			factory.setErrorHandler(new SeekToCurrentErrorHandler());
 			factory.getContainerProperties().setAckMode(AckMode.MANUAL);
 			factory.getContainerProperties().setMissingTopicsFatal(false);
+			factory.setRecordInterceptor(new RecordInterceptor() {
+
+				@Override
+				@Nullable
+				public ConsumerRecord intercept(ConsumerRecord record, Consumer consumer) {
+					return new ConsumerRecord(record.topic(), record.partition(), record.offset(), 0L,
+							TimestampType.NO_TIMESTAMP_TYPE, 0, 0, record.key(), record.value(), record.headers(),
+							Optional.empty());
+				}
+
+			});
 			return factory;
 		}