Allow configuring CompositeBatchInterceptor

This changes allows configuring CompositeBatchInterceptor on AbstractMessageListenerContainer, in same way as CompositeRecordInterceptor.

Signed-off-by: cfredri4 <[email protected]>

Author: cfredri4

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/message-listener-container.adoc

@@ -22,8 +22,8 @@ IMPORTANT: If the interceptor mutates the record (by creating a new one), the `t
 
 The `CompositeRecordInterceptor` and `CompositeBatchInterceptor` can be used to invoke multiple interceptors.
 
-Starting with version 4.0, `AbstractMessageListenerContainer` exposes `getRecordInterceptor()` as a public method.
-If the returned interceptor is an instance of `CompositeRecordInterceptor`, additional `RecordInterceptor` instances can be added to it even after the container instance extending `AbstractMessageListenerContainer` has been created and a `RecordInterceptor` has already been configured.
+Starting with version 4.0, `AbstractMessageListenerContainer` exposes `getRecordInterceptor()` and `getBatchInterceptor()` as public methods.
+If the returned interceptor is an instance of `CompositeRecordInterceptor` or `CompositeBatchInterceptor`, additional `RecordInterceptor` or `BatchInterceptor` instances can be added to it even after the container instance extending `AbstractMessageListenerContainer` has been created and a `RecordInterceptor` has already been configured.
 The following example shows how to do so:
 
 [source, java]

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

@@ -72,6 +72,7 @@
  * @author Soby Chacko
  * @author Sanghyeok An
  * @author Lokesh Alamuri
+ * @author Christian Fredriksson
  */
 public abstract class AbstractMessageListenerContainer<K, V>
 		implements GenericMessageListenerContainer<K, V>, BeanNameAware, ApplicationEventPublisherAware,
@@ -480,7 +481,12 @@ public void setRecordInterceptor(@Nullable RecordInterceptor<K, V> recordInterce
 		this.recordInterceptor = recordInterceptor;
 	}
 
-	protected @Nullable BatchInterceptor<K, V> getBatchInterceptor() {
+	/**
+	 * Get the {@link BatchInterceptor} for modification, if configured.
+	 * @return the {@link BatchInterceptor}, or {@code null} if not configured
+	 * @since 4.0
+	 */
+	public @Nullable BatchInterceptor<K, V> getBatchInterceptor() {
 		return this.batchInterceptor;
 	}
 

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

@@ -34,6 +34,7 @@
  * @param <V> the value type.
  *
  * @author Gary Russell
+ * @author Christian Fredriksson
  * @since 2.7
  *
  */
@@ -85,4 +86,13 @@ public void clearThreadState(Consumer<?, ?> consumer) {
 		this.delegates.forEach(del -> del.clearThreadState(consumer));
 	}
 
+	/**
+	 * Add an {@link BatchInterceptor} to delegates.
+	 * @param batchInterceptor the interceptor.
+	 * @since 4.0
+	 */
+	public void addRecordInterceptor(BatchInterceptor<K, V> batchInterceptor) {
+		this.delegates.add(batchInterceptor);
+	}
+
 }