GH-1587: Don't Correct TX Offsets After Seek(s)

Resolves #1587

We should not advance the consumer partition if Seek operations have been
performed; skip fixing the offsets if that condition is detected.

Capture the positions after the poll and check during the fix operation.

This is difficult to write a unit test for; tested with a Boot application,
observing the correct behavior via DEBUG logs.

**cherry-pick to 2.5.x**

Author: garyrussell

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

@@ -458,6 +458,8 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private final Map<TopicPartition, OffsetAndMetadata> lastCommits = new HashMap<>();
 
+		private final Map<TopicPartition, Long> savedPositions = new HashMap<>();
+
 		private final GenericMessageListener<?> genericListener;
 
 		private final ConsumerSeekAware consumerSeekAwareListener;
@@ -1119,6 +1121,7 @@ protected void pollAndInvoke() {
 			resumeConsumerIfNeccessary();
 			debugRecords(records);
 			if (records != null && records.count() > 0) {
+				savePositionsIfNeeded(records);
 				if (this.containerProperties.getIdleEventInterval() != null) {
 					this.lastReceive = System.currentTimeMillis();
 				}
@@ -1129,13 +1132,28 @@ protected void pollAndInvoke() {
 			}
 		}
 
+		private void savePositionsIfNeeded(ConsumerRecords<K, V> records) {
+			if (this.fixTxOffsets) {
+				this.savedPositions.clear();
+				records.partitions().forEach(tp -> this.savedPositions.put(tp, this.consumer.position(tp)));
+			}
+		}
+
 		@SuppressWarnings("rawtypes")
 		private void fixTxOffsetsIfNeeded() {
 			if (this.fixTxOffsets) {
 				try {
 					Map<TopicPartition, OffsetAndMetadata> toFix = new HashMap<>();
 					this.lastCommits.forEach((tp, oamd) -> {
 						long position = this.consumer.position(tp);
+						Long saved = this.savedPositions.get(tp);
+						if (saved != null && saved.longValue() != position) {
+							this.logger.debug(() -> "Skipping TX offset correction - seek(s) have been performed; "
+									+ "saved: " + this.savedPositions + ", "
+									+ "comitted: " + oamd + ", "
+									+ "current: " + tp + "@" + position);
+							return;
+						}
 						if (position > oamd.offset()) {
 							toFix.put(tp, new OffsetAndMetadata(position));
 						}