GH-1142: Add CompositeRecordInterceptor (#1143)

* GH-1142: Add CompositeRecordInterceptor

Resolves #1142

* Polishing

* * Add assertions

* * Fix checkstyle and varargs warning

Author: garyrussell

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

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.listener;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
+import org.springframework.util.Assert;
+
+/**
+ * A {@link RecordInterceptor} that delegates to one or more {@link RecordInterceptor} in
+ * order.
+ *
+ * @param <K> the key type.
+ * @param <V> the value type.
+ *
+ * @author Artem Bilan
+ * @author Gary Russell
+ * @since 2.3
+ *
+ */
+public class CompositeRecordInterceptor<K, V> implements RecordInterceptor<K, V> {
+
+	private final Collection<RecordInterceptor<K, V>> delegates = new ArrayList<>();
+
+	@SafeVarargs
+	@SuppressWarnings("varargs")
+	public CompositeRecordInterceptor(RecordInterceptor<K, V>... delegates) {
+		Assert.notNull(delegates, "'delegates' cannot be null");
+		Assert.noNullElements(delegates, "'delegates' cannot have null entries");
+		this.delegates.addAll(Arrays.asList(delegates));
+	}
+
+	@Override
+	public ConsumerRecord<K, V> intercept(ConsumerRecord<K, V> record) {
+		ConsumerRecord<K, V> recordToIntercept = record;
+		for (RecordInterceptor<K, V> delegate : this.delegates) {
+			recordToIntercept = delegate.intercept(recordToIntercept);
+		}
+		return recordToIntercept;
+	}
+
+}

src/reference/asciidoc/kafka.adoc

@@ -772,10 +772,12 @@ Two `MessageListenerContainer` implementations are provided:
 The `KafkaMessageListenerContainer` receives all message from all topics or partitions on a single thread.
 The `ConcurrentMessageListenerContainer` delegates to one or more `KafkaMessageListenerContainer` instances to provide multi-threaded consumption.
 
-Starting with version 2.1.7, you can add a `RecordInterceptor` to the listener container; it will be invoked before calling the listener allowing inspection or modification of the record.
+Starting with version 2.2.7, you can add a `RecordInterceptor` to the listener container; it will be invoked before calling the listener allowing inspection or modification of the record.
 If the interceptor returns null, the listener is not called.
 The interceptor is not invoked when the listener is a <<batch-listners, batch listener>>.
 
+Starting with version 2.3, the `CompositeRecordInterceptor` can be used to invoke multiple interceptors.
+
 [[kafka-container]]
 ====== Using `KafkaMessageListenerContainer`
 

src/reference/asciidoc/whats-new.adoc

@@ -39,6 +39,7 @@ It is now possible to obtain the consumer's `group.id` property in the listener
 See <<listener-group-id>> for more information.
 
 The container has a new property `recordInterceptor` allowing records to be inspected or modified before invoking the listener.
+A `CompositeRecordInterceptor` is also provided in case you need to invoke multiple interceptors.
 See <<message-listener-container>> for more information.
 
 ==== ErrorHandler Changes