GH-1569: Add Message Listener Advice Chain

Resolves #1569

Allow provision of around advices, for example to set/reset logging MDC.

**cherry-pick to 2.5.x (what's new will need fixing)**

# Conflicts:
#	src/reference/asciidoc/whats-new.adoc

Author: garyrussell

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

@@ -17,12 +17,20 @@
 package org.springframework.kafka.listener;
 
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.regex.Pattern;
 
+import org.aopalliance.aop.Advice;
+
+import org.springframework.aop.framework.Advised;
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.aop.support.AopUtils;
 import org.springframework.core.task.AsyncListenableTaskExecutor;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.TopicPartitionOffset;
@@ -31,6 +39,7 @@
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.TransactionDefinition;
 import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
 
 /**
  * Contains runtime properties for a listener container.
@@ -164,6 +173,8 @@ public enum EOSMode {
 
 	private final Map<String, String> micrometerTags = new HashMap<>();
 
+	private final List<Advice> adviceChain = new ArrayList<>();
+
 	/**
 	 * The ack mode to use when auto ack (in the configuration properties) is false.
 	 * <ul>
@@ -280,6 +291,7 @@ public ContainerProperties(TopicPartitionOffset... topicPartitions) {
 	 */
 	public void setMessageListener(Object messageListener) {
 		this.messageListener = messageListener;
+		adviseListenerIfNeeded();
 	}
 
 	/**
@@ -709,6 +721,45 @@ public void setTransactionDefinition(TransactionDefinition transactionDefinition
 		this.transactionDefinition = transactionDefinition;
 	}
 
+	/**
+	 * A chain of listener {@link Advice}s.
+	 * @return the adviceChain.
+	 * @since 2.5.6
+	 */
+	public Advice[] getAdviceChain() {
+		return this.adviceChain.toArray(new Advice[0]);
+	}
+
+	/**
+	 * Set a chain of listener {@link Advice}s; must not be null or have null elements.
+	 * @param adviceChain the adviceChain to set.
+	 * @since 2.5.6
+	 */
+	public void setAdviceChain(Advice... adviceChain) {
+		Assert.notNull(adviceChain, "'adviceChain' cannot be null");
+		Assert.noNullElements(adviceChain, "'adviceChain' cannot have null elements");
+		this.adviceChain.clear();
+		this.adviceChain.addAll(Arrays.asList(adviceChain));
+		if (this.messageListener != null) {
+			adviseListenerIfNeeded();
+		}
+	}
+
+	private void adviseListenerIfNeeded() {
+		if (!CollectionUtils.isEmpty(this.adviceChain)) {
+			if (AopUtils.isAopProxy(this.messageListener)) {
+				Advised advised = (Advised) this.messageListener;
+				this.adviceChain.forEach(advised::removeAdvice);
+				this.adviceChain.forEach(advised::addAdvice);
+			}
+			else {
+				ProxyFactory pf = new ProxyFactory(this.messageListener);
+				this.adviceChain.forEach(pf::addAdvice);
+				this.messageListener = pf.getProxy();
+			}
+		}
+	}
+
 	@Override
 	public String toString() {
 		return "ContainerProperties ["

spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

@@ -57,6 +57,7 @@
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
+import org.aopalliance.intercept.MethodInterceptor;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.CommitFailedException;
 import org.apache.kafka.clients.consumer.Consumer;
@@ -596,6 +597,16 @@ public void testRecordAckMock() throws Exception {
 		containerProps.setGroupId("grp");
 		containerProps.setAckMode(AckMode.RECORD);
 		containerProps.setMissingTopicsFatal(false);
+		List<String> advised = new ArrayList<>();
+		MethodInterceptor advice1 = invoc -> {
+			advised.add("one");
+			return invoc.proceed();
+		};
+		MethodInterceptor advice2 = invoc -> {
+			advised.add("two");
+			return invoc.proceed();
+		};
+		containerProps.setAdviceChain(advice1, advice2);
 		final CountDownLatch latch = new CountDownLatch(2);
 		MessageListener<Integer, String> messageListener = spy(
 				new MessageListener<Integer, String>() { // Cannot be lambda: Mockito doesn't mock final classes
@@ -632,6 +643,7 @@ public void onMessage(ConsumerRecord<Integer, String> data) {
 		inOrder.verify(messageListener).onMessage(any(ConsumerRecord.class));
 		inOrder.verify(consumer).commitSync(anyMap(), any());
 		container.stop();
+		assertThat(advised).containsExactly("one", "two", "one", "two");
 	}
 
 	@SuppressWarnings("unchecked")

src/reference/asciidoc/kafka.adoc

@@ -2173,6 +2173,10 @@ To revert to the previous behavior, set the error handler's `resetStateOnRecover
 |1
 |The number of records before committing pending offsets when the `ackMode` is `COUNT` or `COUNT_TIME`.
 
+|adviceChain
+|`null`
+|A chain of `Advice` objects (e.g. `MethodInterceptor` around advice) wrapping the message listener, invoked in order.
+
 |ackMode
 |BATCH
 |Controls how often offsets are committed - see <<committing-offsets>>.