GH-2722: Only Consider interceptBeforeTx with TX

Resolves #2722

Prior to 2.8.x, `interceptBeforeTx` was default `false`; it is now `true`.
However, it should only be considered if there is a `TransactionManager`
present.

This allows an interceptor (when no transactions) to throw an exception,
causing error handling to be invoked.

Also tested with user's reproducer.

**cherry-pick to 2.9.x**

Author: garyrussell

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -37,6 +37,8 @@ public interface BatchInterceptor<K, V> extends ThreadStateProcessor {
 	/**
 	 * Perform some action on the records or return a different one. If null is returned
 	 * the records will be skipped. Invoked before the listener.
+	 * IMPORTANT: If transactions are being used, and this method throws an exception, it
+	 * cannot be used with the container's {@code interceptBeforeTx} property set to true.
 	 * @param records the records.
 	 * @param consumer the consumer.
 	 * @return the records or null.

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

@@ -714,24 +714,24 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 		private final Duration syncCommitTimeout;
 
 		private final RecordInterceptor<K, V> recordInterceptor =
-				!isInterceptBeforeTx() && this.transactionManager != null
+				!isInterceptBeforeTx() || this.transactionManager == null
 						? getRecordInterceptor()
 						: null;
 
 		private final RecordInterceptor<K, V> earlyRecordInterceptor =
-				isInterceptBeforeTx() || this.transactionManager == null
+				isInterceptBeforeTx() && this.transactionManager != null
 						? getRecordInterceptor()
 						: null;
 
 		private final RecordInterceptor<K, V> commonRecordInterceptor = getRecordInterceptor();
 
 		private final BatchInterceptor<K, V> batchInterceptor =
-				!isInterceptBeforeTx() && this.transactionManager != null
+				!isInterceptBeforeTx() || this.transactionManager == null
 						? getBatchInterceptor()
 						: null;
 
 		private final BatchInterceptor<K, V> earlyBatchInterceptor =
-				isInterceptBeforeTx() || this.transactionManager == null
+				isInterceptBeforeTx() && this.transactionManager != null
 						? getBatchInterceptor()
 						: null;
 
@@ -2915,7 +2915,6 @@ private void doInvokeOnMessage(final ConsumerRecord<K, V> recordArg) {
 			if (cRecord == null) {
 				this.logger.debug(() -> "RecordInterceptor returned null, skipping: "
 						+ KafkaUtils.format(recordArg));
-				ackCurrent(recordArg);
 			}
 			else {
 				try {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2022 the original author or authors.
+ * Copyright 2019-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -38,8 +38,11 @@ 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. IMPORTANT; if this method
-	 * returns a different record, the topic, partition and offset must not be changed
-	 * to avoid undesirable side-effects.
+	 * returns a different record, the topic, partition and offset must not be changed to
+	 * avoid undesirable side-effects.
+	 * <p>
+	 * IMPORTANT: If transactions are being used, and this method throws an exception, it
+	 * cannot be used with the container's {@code interceptBeforeTx} property set to true.
 	 * @param record the record.
 	 * @param consumer the consumer.
 	 * @return the record or null.