GH-1455: AdviceChain on Stream Listener Container

Resolves https://github.com/spring-projects/spring-amqp/issues/1455

Add an advice chain to the stream listener container and its factory.
Add a `StreamMessageRecoverer` for native stream messages.
Add a retry interceptor to work with native stream messages.

**cherry-pick to 2.4.x**

* Add since to new setter.

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/config/StreamRabbitListenerContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 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.
@@ -18,12 +18,15 @@ package org.springframework.rabbit.stream.config;
 
 import java.lang.reflect.Method;
 
+import org.aopalliance.aop.Advice;
+
 import org.springframework.amqp.rabbit.batch.BatchingStrategy;
 import org.springframework.amqp.rabbit.config.BaseRabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.config.ContainerCustomizer;
 import org.springframework.amqp.rabbit.listener.MethodRabbitListenerEndpoint;
 import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
 import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
+import org.springframework.amqp.utils.JavaUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.rabbit.stream.listener.ConsumerCustomizer;
 import org.springframework.rabbit.stream.listener.StreamListenerContainer;
@@ -96,9 +99,10 @@ public class StreamRabbitListenerContainerFactory
 							});
 		}
 		StreamListenerContainer container = createContainerInstance();
-		if (this.consumerCustomizer != null) {
-			container.setConsumerCustomizer(this.consumerCustomizer);
-		}
+		Advice[] adviceChain = getAdviceChain();
+		JavaUtils.INSTANCE
+				.acceptIfNotNull(this.consumerCustomizer, container::setConsumerCustomizer)
+				.acceptIfNotNull(adviceChain, container::setAdviceChain);
 		applyCommonOverrides(endpoint, container);
 		if (this.containerCustomizer != null) {
 			this.containerCustomizer.configure(container);

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamListenerContainer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.rabbit.stream.listener;
 
+import org.aopalliance.aop.Advice;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -23,6 +24,8 @@ import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageListener;
 import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.aop.support.DefaultPointcutAdvisor;
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.lang.Nullable;
 import org.springframework.rabbit.stream.support.StreamMessageProperties;
@@ -62,6 +65,10 @@ public class StreamListenerContainer implements MessageListenerContainer, BeanNa
 
 	private MessageListener messageListener;
 
+	private StreamMessageListener streamListener;
+
+	private Advice[] adviceChain;
+
 	/**
 	 * Construct an instance using the provided environment.
 	 * @param environment the environment.
@@ -154,6 +161,18 @@ public class StreamListenerContainer implements MessageListenerContainer, BeanNa
 	public boolean isAutoStartup() {
 		return this.autoStartup;
 	}
+
+	/**
+	 * Set an advice chain to apply to the listener.
+	 * @param advices the advice chain.
+	 * @since 2.4.5
+	 */
+	public void setAdviceChain(Advice... advices) {
+		Assert.notNull(advices, "'advices' cannot be null");
+		Assert.noNullElements(advices, "'advices' cannot have null elements");
+		this.adviceChain = advices;
+	}
+
 	@Override
 	@Nullable
 	public Object getMessageListener() {
@@ -183,26 +202,46 @@ public class StreamListenerContainer implements MessageListenerContainer, BeanNa
 
 	@Override
 	public void setupMessageListener(MessageListener messageListener) {
-		this.messageListener = messageListener;
+		adviseIfNeeded(messageListener);
 		this.builder.messageHandler((context, message) -> {
-			if (messageListener instanceof StreamMessageListener) {
-				((StreamMessageListener) messageListener).onStreamMessage(message, context);
+			if (this.streamListener != null) {
+				this.streamListener.onStreamMessage(message, context);
 			}
 			else {
 				Message message2 = this.streamConverter.toMessage(message, new StreamMessageProperties(context));
-				if (messageListener instanceof ChannelAwareMessageListener) {
+				if (this.messageListener instanceof ChannelAwareMessageListener) {
 					try {
-						((ChannelAwareMessageListener) messageListener).onMessage(message2, null);
+						((ChannelAwareMessageListener) this.messageListener).onMessage(message2, null);
 					}
 					catch (Exception e) { // NOSONAR
 						this.logger.error("Listner threw an exception", e);
 					}
 				}
 				else {
-					messageListener.onMessage(message2);
+					this.messageListener.onMessage(message2);
 				}
 			}
 		});
 	}
 
+	private void adviseIfNeeded(MessageListener messageListener) {
+		this.messageListener = messageListener;
+		if (messageListener instanceof StreamMessageListener) {
+			this.streamListener = (StreamMessageListener) messageListener;
+		}
+		if (this.adviceChain != null && this.adviceChain.length > 0) {
+			ProxyFactory factory = new ProxyFactory(messageListener);
+			for (Advice advice : this.adviceChain) {
+				factory.addAdvisor(new DefaultPointcutAdvisor(advice));
+			}
+			factory.setInterfaces(messageListener.getClass().getInterfaces());
+			if (this.streamListener != null) {
+				this.streamListener = (StreamMessageListener) factory.getProxy(getClass().getClassLoader());
+			}
+			else {
+				this.messageListener = (MessageListener) factory.getProxy(getClass().getClassLoader());
+			}
+		}
+	}
+
 }

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/adapter/StreamMessageListenerAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 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.
@@ -21,6 +21,7 @@ import java.lang.reflect.Method;
 import org.springframework.amqp.rabbit.listener.adapter.InvocationResult;
 import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter;
 import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
+import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException;
 import org.springframework.rabbit.stream.listener.StreamMessageListener;
 
 import com.rabbitmq.stream.Message;
@@ -60,7 +61,7 @@ public class StreamMessageListenerAdapter extends MessagingMessageListenerAdapte
 			}
 		}
 		catch (Exception ex) {
-			this.logger.error("Failed to invoke listener", ex);
+			throw new ListenerExecutionFailedException("Failed to invoke listener", ex);
 		}
 	}
 

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/retry/StreamMessageRecoverer.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2022 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.rabbit.stream.retry;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.retry.MessageRecoverer;
+
+import com.rabbitmq.stream.MessageHandler.Context;
+
+/**
+ * Implementations of this interface can handle failed messages after retries are
+ * exhausted.
+ *
+ * @author Gary Russell
+ * @since 2.4.5
+ *
+ */
+@FunctionalInterface
+public interface StreamMessageRecoverer extends MessageRecoverer {
+
+	@Override
+	default void recover(Message message, Throwable cause) {
+	}
+
+	/**
+	 * Callback for message that was consumed but failed all retry attempts.
+	 *
+	 * @param message the message to recover.
+	 * @param context the context.
+	 * @param cause the cause of the error.
+	 */
+	void recover(com.rabbitmq.stream.Message message, Context context, Throwable cause);
+
+}

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/retry/StreamRetryOperationsInterceptorFactoryBean.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2022 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.rabbit.stream.retry;
+
+import org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptorFactoryBean;
+import org.springframework.amqp.rabbit.retry.MessageRecoverer;
+import org.springframework.rabbit.stream.listener.StreamListenerContainer;
+import org.springframework.retry.RetryOperations;
+import org.springframework.retry.interceptor.MethodInvocationRecoverer;
+import org.springframework.retry.support.RetryTemplate;
+
+import com.rabbitmq.stream.Message;
+import com.rabbitmq.stream.MessageHandler.Context;
+
+/**
+ * Convenient factory bean for creating a stateless retry interceptor for use in a
+ * {@link StreamListenerContainer} when consuming native stream messages, giving you a
+ * large amount of control over the behavior of a container when a listener fails. To
+ * control the number of retry attempt or the backoff in between attempts, supply a
+ * customized {@link RetryTemplate}. Stateless retry is appropriate if your listener can
+ * be called repeatedly between failures with no side effects. The semantics of stateless
+ * retry mean that a listener exception is not propagated to the container until the retry
+ * attempts are exhausted. When the retry attempts are exhausted it can be processed using
+ * a {@link StreamMessageRecoverer} if one is provided.
+ *
+ * @author Gary Russell
+ *
+ * @see RetryOperations#execute(org.springframework.retry.RetryCallback,org.springframework.retry.RecoveryCallback)
+ */
+public class StreamRetryOperationsInterceptorFactoryBean extends StatelessRetryOperationsInterceptorFactoryBean {
+
+	@Override
+	protected MethodInvocationRecoverer<?> createRecoverer() {
+		return (args, cause) -> {
+			StreamMessageRecoverer messageRecoverer = (StreamMessageRecoverer) getMessageRecoverer();
+			Object arg = args[0];
+			if (arg instanceof org.springframework.amqp.core.Message) {
+				return super.recover(args, cause);
+			}
+			else {
+				if (messageRecoverer == null) {
+					this.logger.warn("Message(s) dropped on recovery: " + arg, cause);
+				}
+				else {
+					messageRecoverer.recover((Message) arg, (Context) args[1], cause);
+				}
+				return null;
+			}
+		};
+	}
+
+	/**
+	 * Set a {@link StreamMessageRecoverer} to call when retries are exhausted.
+	 * @param messageRecoverer the recoverer.
+	 */
+	public void setStreamMessageRecoverer(StreamMessageRecoverer messageRecoverer) {
+		super.setMessageRecoverer(messageRecoverer);
+	}
+
+	@Override
+	public void setMessageRecoverer(MessageRecoverer messageRecoverer) {
+		throw new UnsupportedOperationException("Use setStreamMessageRecoverer() instead");
+	}
+
+}

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/retry/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides classes supporting retries.
+ */
+package org.springframework.rabbit.stream.retry;

spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 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.
@@ -23,6 +23,7 @@ import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.junit.jupiter.api.Test;
 
@@ -30,6 +31,7 @@ import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.core.QueueBuilder;
 import org.springframework.amqp.rabbit.annotation.EnableRabbit;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
@@ -38,9 +40,12 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.DependsOn;
 import org.springframework.rabbit.stream.config.StreamRabbitListenerContainerFactory;
 import org.springframework.rabbit.stream.producer.RabbitStreamTemplate;
+import org.springframework.rabbit.stream.retry.StreamRetryOperationsInterceptorFactoryBean;
 import org.springframework.rabbit.stream.support.StreamMessageProperties;
+import org.springframework.retry.interceptor.RetryOperationsInterceptor;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
@@ -64,15 +69,6 @@ public class RabbitListenerTests extends AbstractIntegrationTests {
 	@Autowired
 	Config config;
 
-//	@AfterAll - causes test to throw errors - need to investigate
-	static void deleteQueues() {
-		try (Environment environment = Config.environment()) {
-			environment.deleteStream("test.stream.queue1");
-			environment.deleteStream("test.stream.queue2");
-			environment.deleteStream("stream.created.over.amqp");
-		}
-	}
-
 	@Test
 	void simple(@Autowired RabbitStreamTemplate template) throws Exception {
 		Future<Boolean> future = template.convertAndSend("foo");
@@ -87,8 +83,8 @@ public class RabbitListenerTests extends AbstractIntegrationTests {
 		future = template.convertAndSend("bar", msg -> null);
 		assertThat(future.get(10, TimeUnit.SECONDS)).isFalse();
 		assertThat(this.config.latch1.await(10, TimeUnit.SECONDS)).isTrue();
-		assertThat(this.config.received).containsExactly("foo", "bar", "baz", "qux");
-		assertThat(this.config.id).isEqualTo("test");
+		assertThat(this.config.received).containsExactly("foo", "foo", "bar", "baz", "qux");
+		assertThat(this.config.id).isEqualTo("testNative");
 	}
 
 	@Test
@@ -97,6 +93,8 @@ public class RabbitListenerTests extends AbstractIntegrationTests {
 		assertThat(this.config.latch2.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(this.config.receivedNative).isNotNull();
 		assertThat(this.config.context).isNotNull();
+		assertThat(this.config.latch3.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.config.latch4.await(10, TimeUnit.SECONDS)).isTrue();
 	}
 
 	@Test
@@ -110,12 +108,18 @@ public class RabbitListenerTests extends AbstractIntegrationTests {
 	@EnableRabbit
 	public static class Config {
 
-		final CountDownLatch latch1 = new CountDownLatch(4);
+		final CountDownLatch latch1 = new CountDownLatch(5);
 
 		final CountDownLatch latch2 = new CountDownLatch(1);
 
+		final CountDownLatch latch3 = new CountDownLatch(3);
+
+		final CountDownLatch latch4 = new CountDownLatch(1);
+
 		final List<String> received = new ArrayList<>();
 
+		final AtomicBoolean first = new AtomicBoolean(true);
+
 		volatile Message receivedNative;
 
 		volatile Context context;
@@ -133,12 +137,23 @@ public class RabbitListenerTests extends AbstractIntegrationTests {
 		SmartLifecycle creator(Environment env) {
 			return new SmartLifecycle() {
 
+				boolean running;
+
 				@Override
 				public void stop() {
+					clean(env);
+					this.running = false;
 				}
 
 				@Override
 				public void start() {
+					clean(env);
+					env.streamCreator().stream("test.stream.queue1").create();
+					env.streamCreator().stream("test.stream.queue2").create();
+					this.running = true;
+				}
+
+				private void clean(Environment env) {
 					try {
 						env.deleteStream("test.stream.queue1");
 					}
@@ -149,42 +164,65 @@ public class RabbitListenerTests extends AbstractIntegrationTests {
 					}
 					catch (Exception e) {
 					}
-					env.streamCreator().stream("test.stream.queue1").create();
-					env.streamCreator().stream("test.stream.queue2").create();
+					try {
+						env.deleteStream("stream.created.over.amqp");
+					}
+					catch (Exception e) {
+					}
 				}
 
 				@Override
 				public boolean isRunning() {
-					return false;
+					return this.running;
 				}
 			};
 		}
 
 		@Bean
 		RabbitListenerContainerFactory<StreamListenerContainer> rabbitListenerContainerFactory(Environment env) {
-			return new StreamRabbitListenerContainerFactory(env);
+			StreamRabbitListenerContainerFactory factory = new StreamRabbitListenerContainerFactory(env);
+			factory.setAdviceChain(RetryInterceptorBuilder.stateless().build());
+			return factory;
 		}
 
 		@RabbitListener(queues = "test.stream.queue1")
 		void listen(String in) {
 			this.received.add(in);
 			this.latch1.countDown();
+			if (first.getAndSet(false)) {
+				throw new RuntimeException("fail first");
+			}
 		}
 
 		@Bean
-		RabbitListenerContainerFactory<StreamListenerContainer> nativeFactory(Environment env) {
+		public StreamRetryOperationsInterceptorFactoryBean sfb() {
+			StreamRetryOperationsInterceptorFactoryBean rfb = new StreamRetryOperationsInterceptorFactoryBean();
+			rfb.setStreamMessageRecoverer((msg, context, throwable) -> {
+				this.latch4.countDown();
+			});
+			return rfb;
+		}
+
+		@Bean
+		@DependsOn("sfb")
+		RabbitListenerContainerFactory<StreamListenerContainer> nativeFactory(Environment env,
+				RetryOperationsInterceptor retry) {
+
 			StreamRabbitListenerContainerFactory factory = new StreamRabbitListenerContainerFactory(env);
 			factory.setNativeListener(true);
 			factory.setConsumerCustomizer((id, builder) -> {
-				builder.name("myConsumer")
+				builder.name(id)
 						.offset(OffsetSpecification.first())
 						.manualTrackingStrategy();
-				this.id = id;
+				if (id.equals("testNative")) {
+					this.id = id;
+				}
 			});
+			factory.setAdviceChain(retry);
 			return factory;
 		}
 
-		@RabbitListener(id = "test", queues = "test.stream.queue2", containerFactory = "nativeFactory")
+		@RabbitListener(id = "testNative", queues = "test.stream.queue2", containerFactory = "nativeFactory")
 		void nativeMsg(Message in, Context context) {
 			this.receivedNative = in;
 			this.context = context;
@@ -192,6 +230,12 @@ public class RabbitListenerTests extends AbstractIntegrationTests {
 			context.storeOffset();
 		}
 
+		@RabbitListener(id = "testNativeFail", queues = "test.stream.queue2", containerFactory = "nativeFactory")
+		void nativeMsgFail(Message in, Context context) {
+			this.latch3.countDown();
+			throw new RuntimeException("fail all");
+		}
+
 		@Bean
 		CachingConnectionFactory cf() {
 			return new CachingConnectionFactory("localhost", amqpPort());

spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/StreamListenerContainerTests.java

@@ -0,0 +1,105 @@
+/*
+ * Copyright 2022 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.rabbit.stream.listener;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.willAnswer;
+import static org.mockito.Mockito.mock;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.aopalliance.intercept.MethodInterceptor;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.amqp.core.MessageListener;
+import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
+
+import com.rabbitmq.stream.ConsumerBuilder;
+import com.rabbitmq.stream.Environment;
+import com.rabbitmq.stream.Message;
+import com.rabbitmq.stream.MessageHandler;
+import com.rabbitmq.stream.MessageHandler.Context;
+
+/**
+ * @author Gary Russell
+ * @since 2.4.5
+ *
+ */
+public class StreamListenerContainerTests {
+
+	@Test
+	void testAdviceChain() throws Exception {
+		Environment env = mock(Environment.class);
+		ConsumerBuilder builder = mock(ConsumerBuilder.class);
+		given(env.consumerBuilder()).willReturn(builder);
+		AtomicReference<MessageHandler> handler = new AtomicReference<>();
+		willAnswer(inv -> {
+			handler.set(inv.getArgument(0));
+			return null;
+		}
+		).given(builder).messageHandler(any());
+		AtomicBoolean advised = new AtomicBoolean();
+		MethodInterceptor advice = (inv) -> {
+			advised.set(true);
+			return inv.proceed();
+		};
+
+		StreamListenerContainer container = new StreamListenerContainer(env);
+		container.setAdviceChain(advice);
+		AtomicBoolean called = new AtomicBoolean();
+		MessageListener ml = mock(MessageListener.class);
+		willAnswer(inv -> {
+			called.set(true);
+			return null;
+		}).given(ml).onMessage(any());
+		container.setupMessageListener(ml);
+		Message message = mock(Message.class);
+		given(message.getBodyAsBinary()).willReturn("foo".getBytes());
+		Context context = mock(Context.class);
+		handler.get().handle(context, message);
+		assertThat(advised.get()).isTrue();
+		assertThat(called.get()).isTrue();
+
+		advised.set(false);
+		called.set(false);
+		ChannelAwareMessageListener cal = mock(ChannelAwareMessageListener.class);
+		willAnswer(inv -> {
+			called.set(true);
+			return null;
+		}).given(cal).onMessage(any(), isNull());
+		container.setupMessageListener(cal);
+		handler.get().handle(context, message);
+		assertThat(advised.get()).isTrue();
+		assertThat(called.get()).isTrue();
+
+		called.set(false);
+		StreamMessageListener sml = mock(StreamMessageListener.class);
+		willAnswer(inv -> {
+			called.set(true);
+			return null;
+		}).given(sml).onStreamMessage(message, context);
+		container.setupMessageListener(sml);
+		handler.get().handle(context, message);
+		assertThat(advised.get()).isTrue();
+		assertThat(called.get()).isTrue();
+	}
+
+}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2021 the original author or authors.
+ * Copyright 2014-2022 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.
@@ -39,7 +39,6 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
-import org.springframework.lang.Nullable;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.util.Assert;
 import org.springframework.util.ErrorHandler;
@@ -86,8 +85,6 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 
 	private Boolean globalQos;
 
-	private Advice[] adviceChain;
-
 	private BackOff recoveryBackOff;
 
 	private Boolean missingQueuesFatal;
@@ -182,23 +179,6 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 		this.prefetchCount = prefetch;
 	}
 
-	/**
-	 * @return the advice chain that was set. Defaults to {@code null}.
-	 * @since 1.7.4
-	 */
-	@Nullable
-	public Advice[] getAdviceChain() {
-		return this.adviceChain == null ? null : Arrays.copyOf(this.adviceChain, this.adviceChain.length);
-	}
-
-	/**
-	 * @param adviceChain the advice chain to set.
-	 * @see AbstractMessageListenerContainer#setAdviceChain
-	 */
-	public void setAdviceChain(Advice... adviceChain) {
-		this.adviceChain = adviceChain == null ? null : Arrays.copyOf(adviceChain, adviceChain.length);
-	}
-
 	/**
 	 * @param recoveryInterval The recovery interval.
 	 * @see AbstractMessageListenerContainer#setRecoveryInterval
@@ -354,6 +334,7 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 		if (this.messageConverter != null && endpoint != null && endpoint.getMessageConverter() == null) {
 			endpoint.setMessageConverter(this.messageConverter);
 		}
+		Advice[] adviceChain = getAdviceChain();
 		javaUtils
 			.acceptIfNotNull(this.acknowledgeMode, instance::setAcknowledgeMode)
 			.acceptIfNotNull(this.channelTransacted, instance::setChannelTransacted)
@@ -363,7 +344,7 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 			.acceptIfNotNull(this.prefetchCount, instance::setPrefetchCount)
 			.acceptIfNotNull(this.globalQos, instance::setGlobalQos)
 			.acceptIfNotNull(getDefaultRequeueRejected(), instance::setDefaultRequeueRejected)
-			.acceptIfNotNull(this.adviceChain, instance::setAdviceChain)
+			.acceptIfNotNull(adviceChain, instance::setAdviceChain)
 			.acceptIfNotNull(this.recoveryBackOff, instance::setRecoveryBackOff)
 			.acceptIfNotNull(this.mismatchedQueuesFatal, instance::setMismatchedQueuesFatal)
 			.acceptIfNotNull(this.missingQueuesFatal, instance::setMissingQueuesFatal)

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/BaseRabbitListenerContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 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.
@@ -18,7 +18,10 @@ package org.springframework.amqp.rabbit.config;
 
 import java.util.Arrays;
 
+import org.aopalliance.aop.Advice;
+
 import org.springframework.amqp.core.MessagePostProcessor;
+import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
@@ -49,6 +52,8 @@ public abstract class BaseRabbitListenerContainerFactory<C extends MessageListen
 
 	private RecoveryCallback<?> recoveryCallback;
 
+	private Advice[] adviceChain;
+
 	@Override
 	public abstract C createListenerContainer(RabbitListenerEndpoint endpoint);
 
@@ -129,4 +134,21 @@ public abstract class BaseRabbitListenerContainerFactory<C extends MessageListen
 		}
 	}
 
+	/**
+	 * @return the advice chain that was set. Defaults to {@code null}.
+	 * @since 1.7.4
+	 */
+	@Nullable
+	public Advice[] getAdviceChain() {
+		return this.adviceChain == null ? null : Arrays.copyOf(this.adviceChain, this.adviceChain.length);
+	}
+
+	/**
+	 * @param adviceChain the advice chain to set.
+	 * @see AbstractMessageListenerContainer#setAdviceChain
+	 */
+	public void setAdviceChain(Advice... adviceChain) {
+		this.adviceChain = adviceChain == null ? null : Arrays.copyOf(adviceChain, adviceChain.length);
+	}
+
 }

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/StatelessRetryOperationsInterceptorFactoryBean.java

@@ -46,7 +46,7 @@ import org.springframework.retry.support.RetryTemplate;
  */
 public class StatelessRetryOperationsInterceptorFactoryBean extends AbstractRetryOperationsInterceptorFactoryBean {
 
-	private static Log logger = LogFactory.getLog(StatelessRetryOperationsInterceptorFactoryBean.class);
+	protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR
 
 	@Override
 	public RetryOperationsInterceptor getObject() {
@@ -63,21 +63,23 @@ public class StatelessRetryOperationsInterceptorFactoryBean extends AbstractRetr
 	}
 
 	@SuppressWarnings("unchecked")
-	private MethodInvocationRecoverer<?> createRecoverer() {
-		return (args, cause) -> {
-			MessageRecoverer messageRecoverer = getMessageRecoverer();
-			Object arg = args[1];
-			if (messageRecoverer == null) {
-				logger.warn("Message(s) dropped on recovery: " + arg, cause);
-			}
-			else if (arg instanceof Message) {
-				messageRecoverer.recover((Message) arg, cause);
-			}
-			else if (arg instanceof List && messageRecoverer instanceof MessageBatchRecoverer) {
-				((MessageBatchRecoverer) messageRecoverer).recover((List<Message>) arg, cause);
-			}
-			return null;
-		};
+	protected MethodInvocationRecoverer<?> createRecoverer() {
+		return this::recover;
+	}
+
+	protected Object recover(Object[] args, Throwable cause) {
+		MessageRecoverer messageRecoverer = getMessageRecoverer();
+		Object arg = args[1];
+		if (messageRecoverer == null) {
+			this.logger.warn("Message(s) dropped on recovery: " + arg, cause);
+		}
+		else if (arg instanceof Message) {
+			messageRecoverer.recover((Message) arg, cause);
+		}
+		else if (arg instanceof List && messageRecoverer instanceof MessageBatchRecoverer) {
+			((MessageBatchRecoverer) messageRecoverer).recover((List<Message>) arg, cause);
+		}
+		return null;
 	}
 
 	@Override

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/MessageRecoverer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -19,6 +19,9 @@ package org.springframework.amqp.rabbit.retry;
 import org.springframework.amqp.core.Message;
 
 /**
+ * Implementations of this interface can handle failed messages after retries are
+ * exhausted.
+ *
  * @author Dave Syer
  * @author Gary Russell
  *

src/reference/asciidoc/stream.adoc

@@ -136,3 +136,24 @@ void nativeMsg(Message in, Context context) {
 }
 ----
 ====
+
+Version 2.4.5 added the `adviceChain` property to the `StreamListenerContainer` (and its factory).
+A new factory bean is also provided to create a stateless retry interceptor with an optional `StreamMessageRecoverer` for use when consuming raw stream messages.
+
+====
+[source, java]
+----
+@Bean
+public StreamRetryOperationsInterceptorFactoryBean sfb(RetryTemplate retryTemplate) {
+    StreamRetryOperationsInterceptorFactoryBean rfb =
+            new StreamRetryOperationsInterceptorFactoryBean();
+    rfb.setRetryOperations(retryTemplate);
+    rfb.setStreamMessageRecoverer((msg, context, throwable) -> {
+        ...
+    });
+    return rfb;
+}
+----
+====
+
+IMPORTANT: Stateful retry is not supported with this container.