GH-3542: Adds the ability to add record interceptors instead of override them

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

@@ -460,7 +460,11 @@ public void setKafkaAdmin(KafkaAdmin kafkaAdmin) {
 		this.kafkaAdmin = kafkaAdmin;
 	}
 
-	protected @Nullable RecordInterceptor<K, V> getRecordInterceptor() {
+	/**
+	 * Get the {@link RecordInterceptor} for modification, if configured.
+	 * @return the {@link RecordInterceptor}, or {@code null} if not configured
+	 */
+	public @Nullable RecordInterceptor<K, V> getRecordInterceptor() {
 		return this.recordInterceptor;
 	}
 

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

@@ -35,6 +35,7 @@
  *
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Sanghyeok An
  * @since 2.3
  *
  */
@@ -92,4 +93,13 @@ 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.
+	 * @since 4.0
+	 */
+	public void addRecordInterceptor(RecordInterceptor<K, V> recordInterceptor) {
+		this.delegates.add(recordInterceptor);
+	}
+
 }