Add new API addRecordInterceptor.

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

Author: chickenchickenlove

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

@@ -475,6 +475,24 @@ public void setRecordInterceptor(@Nullable RecordInterceptor<K, V> recordInterce
 		this.recordInterceptor = recordInterceptor;
 	}
 
+	/**
+	 * Add an interceptor to be called before calling the record listener
+	 * if the {@link AbstractMessageListenerContainer} is configured with a {@link CompositeRecordInterceptor}.
+	 * If a {@link CompositeRecordInterceptor} is not configured, the {@link AbstractMessageListenerContainer}
+	 * will not add {@link RecordInterceptor}.
+	 * Does not apply to batch listeners.
+	 * @param recordInterceptor the interceptor.
+	 * @since 4.0
+	 */
+	public void addRecordInterceptor(RecordInterceptor<K, V> recordInterceptor) {
+		if (this.recordInterceptor instanceof CompositeRecordInterceptor<K, V> compositeRecordInterceptor) {
+			compositeRecordInterceptor.addRecordInterceptor(recordInterceptor);
+		}
+		else {
+			this.logger.warn("Failed to add record interceptor.");
+		}
+	}
+
 	protected @Nullable BatchInterceptor<K, V> getBatchInterceptor() {
 		return this.batchInterceptor;
 	}

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

@@ -92,4 +92,12 @@ public void afterRecord(ConsumerRecord<K, V> record, Consumer<K, V> consumer) {
 		this.delegates.forEach(del -> del.afterRecord(record, consumer));
 	}
 
+	/**
+	 * Add an {@link RecordInterceptor} to delegates.
+	 * @param recordInterceptor the interceptor.
+	 */
+	public void addRecordInterceptor(RecordInterceptor<K, V> recordInterceptor) {
+		this.delegates.add(recordInterceptor);
+	}
+
 }