GH-1929: Fix BatchInterceptor

Resolves #1929

Need to rebuild the record list after interception (when `interceptBeforeTx` is false).

Otherwise changes made to the records in the `ConsumerRecords` is lost.

Always intercept early when transactions are not used; rebuild list if transactions in
use, `interceptBeforeTx` is false, and a batch interceptor is present.

**cherry-pick to 2.7.x**

Author: garyrussell

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

@@ -611,23 +611,25 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private final Duration syncCommitTimeout;
 
-		private final RecordInterceptor<K, V> recordInterceptor = !isInterceptBeforeTx()
+		private final RecordInterceptor<K, V> recordInterceptor = !isInterceptBeforeTx() && this.kafkaTxManager != null
 				? getRecordInterceptor()
 				: null;
 
-		private final RecordInterceptor<K, V> earlyRecordInterceptor = isInterceptBeforeTx()
-				? getRecordInterceptor()
-				: null;
+		private final RecordInterceptor<K, V> earlyRecordInterceptor =
+				isInterceptBeforeTx() || this.kafkaTxManager == null
+						? getRecordInterceptor()
+						: null;
 
 		private final RecordInterceptor<K, V> commonRecordInterceptor = getRecordInterceptor();
 
-		private final BatchInterceptor<K, V> batchInterceptor = !isInterceptBeforeTx()
+		private final BatchInterceptor<K, V> batchInterceptor = !isInterceptBeforeTx() && this.kafkaTxManager != null
 				? getBatchInterceptor()
 				: null;
 
-		private final BatchInterceptor<K, V> earlyBatchInterceptor = isInterceptBeforeTx()
-				? getBatchInterceptor()
-				: null;
+		private final BatchInterceptor<K, V> earlyBatchInterceptor =
+				isInterceptBeforeTx() || this.kafkaTxManager == null
+						? getBatchInterceptor()
+						: null;
 
 		private final BatchInterceptor<K, V> commonBatchInterceptor = getBatchInterceptor();
 
@@ -1961,11 +1963,20 @@ private void invokeBatchOnMessage(final ConsumerRecords<K, V> records, // NOSONA
 		}
 
 		private void invokeBatchOnMessageWithRecordsOrList(final ConsumerRecords<K, V> recordsArg,
-				@Nullable List<ConsumerRecord<K, V>> recordList) {
+				@Nullable List<ConsumerRecord<K, V>> recordListArg) {
 
 			ConsumerRecords<K, V> records = recordsArg;
+			List<ConsumerRecord<K, V>> recordList = recordListArg;
 			if (this.batchInterceptor != null) {
 				records = this.batchInterceptor.intercept(recordsArg, this.consumer);
+				if (records == null) {
+					this.logger.debug(() -> "BatchInterceptor returned null, skipping: "
+							+ recordsArg + " with " + recordsArg.count() + " records");
+					return;
+				}
+				else {
+					recordList = createRecordList(records);
+				}
 			}
 			if (this.wantsFullRecords) {
 				this.batchListener.onMessage(records, // NOSONAR
@@ -2154,7 +2165,7 @@ private ConsumerRecords<K, V> checkEarlyIntercept(ConsumerRecords<K, V> nextArg)
 			if (this.earlyBatchInterceptor != null) {
 				next = this.earlyBatchInterceptor.intercept(next, this.consumer);
 				if (next == null) {
-					this.logger.debug(() -> "RecordInterceptor returned null, skipping: "
+					this.logger.debug(() -> "BatchInterceptor returned null, skipping: "
 						+ nextArg + " with " + nextArg.count() + " records");
 				}
 			}

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

@@ -450,8 +450,11 @@ void testIntercept(boolean beforeTx, AssignmentCommitOption option, boolean batc
 		ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
 		final Consumer consumer = mock(Consumer.class);
 		TopicPartition tp0 = new TopicPartition("foo", 0);
-		ConsumerRecord record = new ConsumerRecord("foo", 0, 0L, "bar", "baz");
-		ConsumerRecords records = new ConsumerRecords(Collections.singletonMap(tp0, Collections.singletonList(record)));
+		ConsumerRecord record1 = new ConsumerRecord("foo", 0, 0L, "bar", "baz");
+		ConsumerRecord record2 = new ConsumerRecord("foo", 0, 1L, null, null);
+		ConsumerRecords records = batch
+				? new ConsumerRecords(Collections.singletonMap(tp0, List.of(record1, record2)))
+				: new ConsumerRecords(Collections.singletonMap(tp0, Collections.singletonList(record1)));
 		ConsumerRecords empty = new ConsumerRecords<>(Collections.emptyMap());
 		AtomicInteger firstOrSecondPoll = new AtomicInteger();
 		willAnswer(invocation -> {
@@ -470,11 +473,13 @@ void testIntercept(boolean beforeTx, AssignmentCommitOption option, boolean batc
 		ContainerProperties containerProperties = new ContainerProperties("foo");
 		containerProperties.setGroupId("grp");
 		AtomicBoolean first = new AtomicBoolean(true);
+		AtomicReference<List<ConsumerRecord<String, String>>> received = new AtomicReference<>();
 		if (batch) {
-			containerProperties.setMessageListener((BatchMessageListener) recs -> {
+			containerProperties.setMessageListener((BatchMessageListener<String, String>) recs -> {
 				if (first.getAndSet(false)) {
 					throw new RuntimeException("test");
 				}
+				received.set(recs);
 			});
 		}
 		else {
@@ -546,7 +551,7 @@ public void failure(ConsumerRecord record, Exception exception, Consumer consume
 			public ConsumerRecords intercept(ConsumerRecords recs, Consumer consumer) {
 				order.add("interceptor");
 				latch.countDown();
-				return recs;
+				return new ConsumerRecords(Collections.singletonMap(tp0, Collections.singletonList(record1)));
 			}
 
 			@Override
@@ -586,6 +591,9 @@ public void failure(ConsumerRecords records, Exception exception, Consumer consu
 							"success");
 				}
 			}
+			if (batch) {
+				assertThat(received.get()).hasSize(1);
+			}
 		}
 		finally {
 			container.stop();