Refactor fixTxOffsets to use ConsumerRecords#nextOffsets

Replaced the position()-based offset correction with nextOffsets() to
address known issues with transactional topics:
 - Works correctly with both read_committed and read_uncommitted
 - Handles empty poll() advancement and leader epoch propagation
 - Fixes the case when max.poll.records equals transaction batch size

Signed-off-by: Su Ko <[email protected]>

Author: bandalgomsu

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

@@ -156,7 +156,6 @@ public class TransactionalContainerTests {
 
 	public static final String topic15 = "txTopic15";
 
-
 	private static EmbeddedKafkaBroker embeddedKafka;
 
 	@BeforeAll
@@ -1323,7 +1322,8 @@ void testFixTxOffsetsWithEmptyPollAdvance() throws Exception {
 		containerProps.setFixTxOffsets(true);
 		containerProps.setIdleEventInterval(1000L);
 
-		containerProps.setMessageListener((MessageListener<Integer, String>) rec -> {});
+		containerProps.setMessageListener((MessageListener<Integer, String>) rec -> {
+		});
 
 		DefaultKafkaProducerFactory<Integer, String> pf =
 				new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(embeddedKafka));
@@ -1371,7 +1371,8 @@ void testFixTxOffsetsRetainsLeaderEpoch() throws Exception {
 		containerProps.setFixTxOffsets(true);
 		containerProps.setIdleEventInterval(1000L);
 
-		containerProps.setMessageListener((MessageListener<Integer, String>) rec -> {});
+		containerProps.setMessageListener((MessageListener<Integer, String>) rec -> {
+		});
 
 		DefaultKafkaProducerFactory<Integer, String> pf =
 				new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(embeddedKafka));
@@ -1417,7 +1418,8 @@ void testFixLagWhenMaxPollEqualsTxBatchSize() throws Exception {
 		containerProps.setPollTimeout(500L);
 		containerProps.setFixTxOffsets(true);
 		containerProps.setIdleEventInterval(1000L);
-		containerProps.setMessageListener((MessageListener<Integer, String>) rec -> {});
+		containerProps.setMessageListener((MessageListener<Integer, String>) rec -> {
+		});
 
 		DefaultKafkaProducerFactory<Integer, String> pf =
 				new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(embeddedKafka));