Configure interceptors on container factory

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

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,32 +22,39 @@ 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` 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:
+The following example shows how to do so in a `BeanPostProcessor`:
 
 [source, java]
 ----
-public void configureRecordInterceptor(KafkaMessageListenerContainer<Integer, String> container) {
-    CompositeRecordInterceptor compositeInterceptor;
-
-    RecordInterceptor<Integer, String> previousInterceptor = container.getRecordInterceptor();
-    if (previousInterceptor instanceof CompositeRecordInterceptor interceptor) {
-        compositeInterceptor = interceptor;
-    } else {
-        compositeInterceptor = new CompositeRecordInterceptor<>();
-        container.setRecordInterceptor(compositeInterceptor);
-    }
-
-    if (previousInterceptor != null) {
-        compositeRecordInterceptor.addRecordInterceptor(previousInterceptor);
+public class RecordInterceptorBeanPostProcessor implements BeanPostProcessor {
+
+    @Override
+    public Object postProcessAfterInitialization(Object bean, String beanName) {
+        if (bean instanceof AbstractKafkaListenerContainerFactory containerFactory) {
+
+            RecordInterceptor previousInterceptor = containerFactory.getRecordInterceptor();
+
+            CompositeRecordInterceptor compositeInterceptor;
+            if (previousInterceptor instanceof CompositeRecordInterceptor interceptor) {
+                compositeInterceptor = interceptor;
+            } else {
+                compositeInterceptor = new CompositeRecordInterceptor<>();
+                containerFactory.setRecordInterceptor(compositeInterceptor);
+                if (previousInterceptor != null) {
+                    compositeRecordInterceptor.addRecordInterceptor(previousInterceptor);
+                }
+            }
+
+            RecordInterceptor<Integer, String> recordInterceptor1 = new RecordInterceptor() {...};
+            RecordInterceptor<Integer, String> recordInterceptor2 = new RecordInterceptor() {...};
+
+            compositeInterceptor.addRecordInterceptor(recordInterceptor1);
+            compositeInterceptor.addRecordInterceptor(recordInterceptor2);
+        }
+        return bean;
     }
-
-    RecordInterceptor<Integer, String> recordInterceptor1 = new RecordInterceptor() {...};
-    RecordInterceptor<Integer, String> recordInterceptor2 = new RecordInterceptor() {...};
-
-    compositeInterceptor.addRecordInterceptor(recordInterceptor1);
-    compositeInterceptor.addRecordInterceptor(recordInterceptor2);
 }
 ----
 

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.