Revert 2 commits.

Signed-off-by: Sanghyeok An <[email protected]>

Author: chickenchickenlove

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

@@ -16,7 +16,6 @@
 
 package org.springframework.kafka.listener;
 
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
@@ -115,7 +114,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
 
 	private int topicCheckTimeout = DEFAULT_TOPIC_CHECK_TIMEOUT;
 
-	private List<RecordInterceptor<K, V>> recordInterceptors = new ArrayList<>();
+	private @Nullable RecordInterceptor<K, V> recordInterceptor;
 
 	private @Nullable BatchInterceptor<K, V> batchInterceptor;
 
@@ -461,8 +460,8 @@ public void setKafkaAdmin(KafkaAdmin kafkaAdmin) {
 		this.kafkaAdmin = kafkaAdmin;
 	}
 
-	protected List<RecordInterceptor<K, V>> getRecordInterceptors() {
-		return this.recordInterceptors;
+	protected @Nullable RecordInterceptor<K, V> getRecordInterceptor() {
+		return this.recordInterceptor;
 	}
 
 	/**
@@ -473,9 +472,7 @@ protected List<RecordInterceptor<K, V>> getRecordInterceptors() {
 	 * @see #setInterceptBeforeTx(boolean)
 	 */
 	public void setRecordInterceptor(@Nullable RecordInterceptor<K, V> recordInterceptor) {
-		if (recordInterceptor != null) {
-			this.recordInterceptors.add(recordInterceptor);
-		}
+		this.recordInterceptor = recordInterceptor;
 	}
 
 	protected @Nullable BatchInterceptor<K, V> getBatchInterceptor() {

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

@@ -60,7 +60,6 @@
  * @author Tomaz Fernandes
  * @author Wang Zhiyang
  * @author Lokesh Alamuri
- * @author Sanghyeok An
  */
 public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageListenerContainer<K, V> {
 
@@ -283,9 +282,7 @@ private void configureChildContainer(int index, KafkaMessageListenerContainer<K,
 		container.setClientIdSuffix(this.concurrency > 1 || this.alwaysClientIdSuffix ? "-" + index : "");
 		container.setCommonErrorHandler(getCommonErrorHandler());
 		container.setAfterRollbackProcessor(getAfterRollbackProcessor());
-		for (RecordInterceptor<K, V> recordInterceptor : getRecordInterceptors()) {
-			container.setRecordInterceptor(recordInterceptor);
-		}
+		container.setRecordInterceptor(getRecordInterceptor());
 		container.setBatchInterceptor(getBatchInterceptor());
 		container.setInterceptBeforeTx(isInterceptBeforeTx());
 		container.setListenerInfo(getListenerInfo());

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

@@ -714,17 +714,17 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private final @Nullable Duration syncCommitTimeout;
 
-		private final List<RecordInterceptor<K, V>> recordInterceptors =
+		private final @Nullable RecordInterceptor<K, V> recordInterceptor =
 				!isInterceptBeforeTx() || this.transactionManager == null
-						? getRecordInterceptors()
-						: new ArrayList<>();
+						? getRecordInterceptor()
+						: null;
 
-		private final List<RecordInterceptor<K, V>> earlyRecordInterceptors =
+		private final @Nullable RecordInterceptor<K, V> earlyRecordInterceptor =
 				isInterceptBeforeTx() && this.transactionManager != null
-						? getRecordInterceptors()
-						: new ArrayList<>();
+						? getRecordInterceptor()
+						: null;
 
-		private final List<RecordInterceptor<K, V>> commonRecordInterceptors = getRecordInterceptors();
+		private final @Nullable RecordInterceptor<K, V> commonRecordInterceptor = getRecordInterceptor();
 
 		private final @Nullable BatchInterceptor<K, V> batchInterceptor =
 				!isInterceptBeforeTx() || this.transactionManager == null
@@ -738,7 +738,7 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private final @Nullable BatchInterceptor<K, V> commonBatchInterceptor = getBatchInterceptor();
 
-		private final List<ThreadStateProcessor> pollThreadStateProcessor;
+		private final @Nullable ThreadStateProcessor pollThreadStateProcessor;
 
 		private final ConsumerSeekCallback seekCallback = new InitialOrIdleSeekCallback();
 
@@ -1040,20 +1040,9 @@ private void obtainClusterId() {
 			}
 		}
 
-		private List<ThreadStateProcessor> setUpPollProcessor(boolean batch) {
-			if (batch) {
-				if (this.commonBatchInterceptor != null) {
-					List<ThreadStateProcessor> threadStateProcessors = new ArrayList<>();
-					threadStateProcessors.add(this.commonBatchInterceptor);
-					return threadStateProcessors;
-				}
-				else {
-					return new ArrayList<>();
-				}
-			}
-			else {
-				return new ArrayList<>(this.commonRecordInterceptors);
-			}
+		@Nullable
+		private ThreadStateProcessor setUpPollProcessor(boolean batch) {
+			return batch ? this.commonBatchInterceptor : this.commonRecordInterceptor;
 		}
 
 		@Nullable
@@ -1559,7 +1548,9 @@ private void invokeIfHaveRecords(@Nullable ConsumerRecords<K, V> records) {
 		}
 
 		private void clearThreadState() {
-			this.pollThreadStateProcessor.forEach(threadStateProcessor -> threadStateProcessor.clearThreadState(this.consumer));
+			if (this.pollThreadStateProcessor != null) {
+				this.pollThreadStateProcessor.clearThreadState(this.consumer);
+			}
 		}
 
 		private void checkIdlePartitions() {
@@ -1717,7 +1708,9 @@ private ConsumerRecords<K, V> pollConsumer() {
 		}
 
 		private void beforePoll() {
-			this.pollThreadStateProcessor.forEach(threadStateProcessor -> threadStateProcessor.setupThreadState(this.consumer));
+			if (this.pollThreadStateProcessor != null) {
+				this.pollThreadStateProcessor.setupThreadState(this.consumer);
+			}
 		}
 
 		private synchronized void captureOffsets(ConsumerRecords<K, V> records) {
@@ -2555,7 +2548,9 @@ private void invokeRecordListenerInTx(final ConsumerRecords<K, V> records) {
 					this.logger.error(ex, "Transaction rolled back");
 					recordAfterRollback(iterator, cRecord, ex);
 				}
-				this.commonRecordInterceptors.forEach(interceptor -> interceptor.afterRecord(cRecord, this.consumer));
+				if (this.commonRecordInterceptor != null) {
+					this.commonRecordInterceptor.afterRecord(cRecord, this.consumer);
+				}
 				if (this.nackSleepDurationMillis >= 0) {
 					handleNack(records, cRecord);
 					break;
@@ -2632,7 +2627,9 @@ private void doInvokeWithRecords(final ConsumerRecords<K, V> records) {
 				}
 				this.logger.trace(() -> "Processing " + KafkaUtils.format(cRecord));
 				doInvokeRecordListener(cRecord, iterator);
-				this.commonRecordInterceptors.forEach(interceptor -> interceptor.afterRecord(cRecord, this.consumer));
+				if (this.commonRecordInterceptor !=  null) {
+					this.commonRecordInterceptor.afterRecord(cRecord, this.consumer);
+				}
 				if (this.nackSleepDurationMillis >= 0) {
 					handleNack(records, cRecord);
 					break;
@@ -2683,16 +2680,14 @@ private ConsumerRecords<K, V> checkEarlyIntercept(ConsumerRecords<K, V> nextArg)
 		private ConsumerRecord<K, V> checkEarlyIntercept(ConsumerRecord<K, V> recordArg) {
 			internalHeaders(recordArg);
 			ConsumerRecord<K, V> cRecord = recordArg;
-
-			for (RecordInterceptor<K, V> earlyRecordInterceptor : this.earlyRecordInterceptors) {
-				cRecord = earlyRecordInterceptor.intercept(cRecord, this.consumer);
+			if (this.earlyRecordInterceptor != null) {
+				cRecord = this.earlyRecordInterceptor.intercept(cRecord, this.consumer);
 				if (cRecord == null) {
 					this.logger.debug(() -> "RecordInterceptor returned null, skipping: "
-											+ KafkaUtils.format(recordArg));
+						+ KafkaUtils.format(recordArg));
 					ackCurrent(recordArg);
-					earlyRecordInterceptor.success(recordArg, this.consumer);
-					earlyRecordInterceptor.afterRecord(recordArg, this.consumer);
-					break;
+					this.earlyRecordInterceptor.success(recordArg, this.consumer);
+					this.earlyRecordInterceptor.afterRecord(recordArg, this.consumer);
 				}
 			}
 			return cRecord;
@@ -2853,17 +2848,19 @@ private void commitOffsetsIfNeededAfterHandlingError(final ConsumerRecord<K, V>
 		}
 
 		private void recordInterceptAfter(ConsumerRecord<K, V> records, @Nullable Exception exception) {
-			try {
-				if (exception == null) {
-					this.commonRecordInterceptors.forEach(interceptor -> interceptor.success(records, this.consumer));
+			if (this.commonRecordInterceptor != null) {
+				try {
+					if (exception == null) {
+						this.commonRecordInterceptor.success(records, this.consumer);
+					}
+					else {
+						this.commonRecordInterceptor.failure(records, exception, this.consumer);
+					}
 				}
-				else {
-					this.commonRecordInterceptors.forEach(interceptor -> interceptor.failure(records, exception, this.consumer));
+				catch (Exception e) {
+					this.logger.error(e, "RecordInterceptor.success/failure threw an exception");
 				}
 			}
-			catch (Exception e) {
-				this.logger.error(e, "RecordInterceptor.success/failure threw an exception");
-			}
 		}
 
 		private void invokeOnMessage(final ConsumerRecord<K, V> cRecord) {
@@ -2891,11 +2888,8 @@ private void invokeOnMessage(final ConsumerRecord<K, V> cRecord) {
 
 		private void doInvokeOnMessage(final ConsumerRecord<K, V> recordArg) {
 			ConsumerRecord<K, V> cRecord = recordArg;
-			for (RecordInterceptor<K, V> recordInterceptor : this.recordInterceptors) {
-				cRecord = recordInterceptor.intercept(cRecord, this.consumer);
-				if (cRecord == null) {
-					break;
-				}
+			if (this.recordInterceptor != null) {
+				cRecord = this.recordInterceptor.intercept(cRecord, this.consumer);
 			}
 			if (cRecord == null) {
 				this.logger.debug(() -> "RecordInterceptor returned null, skipping: "