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**

Author: garyrussell

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

@@ -2447,14 +2447,20 @@ 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
 			pauseForNackSleep();
 		}
 
+		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 pauseForNackSleep() {
 			if (this.nackSleep > 0) {
 				this.nackWake = System.currentTimeMillis() + this.nackSleep;

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

@@ -37,7 +37,9 @@ public interface RecordInterceptor<K, V> extends ThreadStateProcessor {
 
 	/**
 	 * 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.
 	 * @param consumer the consumer.
 	 * @return the record or null.

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

@@ -61,6 +61,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;
 
@@ -243,6 +244,17 @@ public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
 			factory.setConsumerFactory(consumerFactory());
 			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;
 		}