Can StreamRabbitListenerContainerFactory message listeners retry messages?

[daniel-bray-sonalake]

Hi, thanks in advance for any help

We're using the StreamRabbitListenerContainerFactory to build our connection listener for our rabbit stream, and it's working fine, but there are some cases where we get timing issues where we know a retried message will eventually process ok.

Our issue is that, as far as I can tell, consumers created by this factory don't support retries

• This factory has an setRetryTemplate operation, and I've set it, but it seems to do nothing
• I've also tried setting an errorHandler in the @RabbitListener but it's not working.

I suspect the problem is that the issue here in StreamMessageListenerAdapter#onStreamMessage where, any exception is just swallowed, so no retry / error logic can be called.

So I'm wondering if this streams container factory just doesn't support retries: is this true?

Is there some other way to process streams that supports retries, and also follows the offset specification?

Thanks

[garyrussell]

The retry template in the listener container factory is for retrying publishing replies (for a request/reply listener).

	/**
	 * Set a {@link RetryTemplate} to use when sending replies; added to each message
	 * listener adapter.
	 * @param retryTemplate the template.
	 * @see #setReplyRecoveryCallback(RecoveryCallback)
	 * @see AbstractAdaptableMessageListener#setRetryTemplate(RetryTemplate)
	 */
	public void setRetryTemplate(RetryTemplate retryTemplate) {
		this.retryTemplate = retryTemplate;
	}

We don't currently support request/reply for stream listeners.

Retries for deliveries in the classic listener containers is implemented using the advice chain (with a retry interceptor https://docs.spring.io/spring-amqp/docs/current/reference/html/#retry); there is currently no advice chain support in the StreamListenerContainer.

We can look into doing that; please open a new feature issue.

In the meantime, you can use a RetryTemplate within your listener method (or using spring-retry's @Retryable annotation on the listener should work too).

[garyrussell]

Here's an example:

@SpringBootApplication
@EnableRetry
public class Rgh1454Application {

	private static final Logger log = LoggerFactory.getLogger(Rgh1454Application.class);

	public static void main(String[] args) {
		SpringApplication.run(Rgh1454Application.class, args);
	}

	@RabbitListener(queues = "fooStream")
	@Retryable(maxAttempts = 3, backoff = @Backoff(delay = 1_000, multiplier = 1.5, maxDelay = 10_000))
	public void listen(String in) {
	 	log.info(in);
		throw new RuntimeException("test");
	}

	@Bean
	ApplicationRunner runner(RabbitTemplate template) {
		return args -> {
			template.convertAndSend("fooStream", "bar");
		};
	}

}

[daniel-bray-sonalake]

We don't currently support request/reply for stream listeners.

Thanks

In the meantime, you can use a RetryTemplate within your listener method

Thanks, I'll check it out.

We can look into doing that; please open a new feature issue.

I'll do this

[daniel-bray-sonalake]

We can look into doing that; please open a new feature issue.

Raised as: #1455

[daniel-bray-sonalake]

In the meantime, you can use a RetryTemplate within your listener method
Thanks, I'll check it out.

Just to clarify - and with thanks - this will do as a workaround, but it's not a safe resolution, because if the spring app is restarted while we're retrying it this way, the message is lost. What we'd be hoping for here is the ability to throw an error so the context offset isn't updated