Configure interceptors on container factory (#4080)

This changes allows configuring interceptors on AbstractKafkaListenerContainerFactory in same way as on KafkaMessageListenerContainer, and changes the example docs to show this instead.

Fixes #4080

Signed-off-by: Christian Fredriksson <[email protected]>

Author: cfredri4

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

@@ -22,25 +22,24 @@ 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()` and `getBatchInterceptor()` as public methods.
+Starting with version 4.0, `AbstractKafkaListenerContainerFactory` and `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` or `BatchInterceptor` has already been configured.
 The following example shows how to do so:
 
 [source, java]
 ----
-public void configureRecordInterceptor(KafkaMessageListenerContainer<Integer, String> container) {
+public void configureRecordInterceptor(AbstractKafkaListenerContainerFactory<Integer, String> containerFactory) {
     CompositeRecordInterceptor compositeInterceptor;
 
-    RecordInterceptor<Integer, String> previousInterceptor = container.getRecordInterceptor();
+    RecordInterceptor<Integer, String> previousInterceptor = containerFactory.getRecordInterceptor();
     if (previousInterceptor instanceof CompositeRecordInterceptor interceptor) {
         compositeInterceptor = interceptor;
     } else {
         compositeInterceptor = new CompositeRecordInterceptor<>();
-        container.setRecordInterceptor(compositeInterceptor);
-    }
-
-    if (previousInterceptor != null) {
-        compositeRecordInterceptor.addRecordInterceptor(previousInterceptor);
+        containerFactory.setRecordInterceptor(compositeInterceptor);
+        if (previousInterceptor != null) {
+            compositeRecordInterceptor.addRecordInterceptor(previousInterceptor);
+        }
     }
 
     RecordInterceptor<Integer, String> recordInterceptor1 = new RecordInterceptor() {...};

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java

@@ -61,6 +61,7 @@
  * @author Stephane Nicoll
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Christian Fredriksson
  *
  * @see AbstractMessageListenerContainer
  */
@@ -275,6 +276,15 @@ public ContainerProperties getContainerProperties() {
 		return this.containerProperties;
 	}
 
+	/**
+	 * Get the {@link RecordInterceptor} for modification, if configured.
+	 * @return the {@link RecordInterceptor}, or {@code null} if not configured
+	 * @since 4.0
+	 */
+	public @Nullable RecordInterceptor<K, V> getRecordInterceptor() {
+		return this.recordInterceptor;
+	}
+
 	/**
 	 * Set an interceptor to be called before calling the listener.
 	 * Only used with record listeners.
@@ -286,6 +296,15 @@ public void setRecordInterceptor(RecordInterceptor<K, V> recordInterceptor) {
 		this.recordInterceptor = recordInterceptor;
 	}
 
+	/**
+	 * 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;
+	}
+
 	/**
 	 * Set a batch interceptor to be called before and after calling the listener.
 	 * Only used with batch listeners.