Queue and exchange recovery for manually created message listener containers

[tex1988]

I'm trying to understand how listener containers created with the @RabbitListener annotation handle queue and exchange recovery. If the queue being listened to is deleted or becomes unavailable, Spring automatically recovers it and restarts the container when the queue becomes available again.
However, when I create a container manually using the following code:

private void createMessageListenerContainer(Queue queue) {
        MethodRabbitListenerEndpoint endpoint = new MethodRabbitListenerEndpoint();
        RabbitRpcValidator rpcValidator = new RabbitRpcValidator(validator, getServiceName());
        RabbitRpcMessageHandler handler = new RabbitRpcMessageHandler(rpcValidator, messageConverter, methodHandles);
        Method handleMethod = handler.getClass().getMethod(HANDLER_METHOD_NAME, Message.class, Channel.class, MessageProperties.class);
        endpoint.setId(queue.getName() + "-" + getServiceName());
        endpoint.setQueues(queue);
        endpoint.setBean(handler);
        endpoint.setMethod(handleMethod);
        endpoint.setErrorHandler(errorHandler);
        endpoint.setMessageHandlerMethodFactory(messageHandlerMethodFactory);
        endpoint.setBeanFactory(applicationContext);
        endpoint.setAdmin(amqpAdmin);
        RabbitListenerEndpointRegistry registry = applicationContext.getBean(
                RabbitListenerConfigUtils.RABBIT_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,
                RabbitListenerEndpointRegistry.class);
        registry.registerListenerContainer(endpoint, rabbitListenerContainerFactory);
    }

the container is restarted, but it fails with an exception indicating that the queue is not found.

rabbitListenerContainerFactory is configured by code:

rabbitListenerContainerFactory = new SimpleRabbitListenerContainerFactory();
configurer.configure(rabbitListenerContainerFactory, connectionFactory);
rabbitListenerContainerFactory.setMessageConverter(messageConverter);
rabbitListenerContainerFactory.setFailedDeclarationRetryInterval(10000L);
rabbitListenerContainerFactory.setMissingQueuesFatal(true);

Where configurer and connectionFactory beans created by spring-boot-starter-amqp
I would appreciate any help on how to make Spring take care of the queues and exchanges used by my custom listener containers.

[artembilan]

See docs for Broker entities configuration: https://docs.spring.io/spring-amqp/reference/amqp/broker-configuration.html.

By default only those exchanges, queues, and binding between them are re-declared on re-connection which are beans in the application context.

It is not clear what is your custom listener container, but you can use RabbitAdmin API manually to declare those entities.
Plus pay attention to the redeclareManualDeclarations property: https://docs.spring.io/spring-amqp/reference/amqp/broker-configuration.html

[tex1988]

See docs for Broker entities configuration: https://docs.spring.io/spring-amqp/reference/amqp/broker-configuration.html.

By default only those exchanges, queues, and binding between them are re-declared on re-connection which are beans in the application context.

It is not clear what is your custom listener container, but you can use RabbitAdmin API manually to declare those entities. Plus pay attention to the redeclareManualDeclarations property: https://docs.spring.io/spring-amqp/reference/amqp/broker-configuration.html

I've created beans for a queue and an exchange without attaching any listener. Then I manually deleted the queue via the RabbitMQ UI, but it wasn't redeclared automatically. It seems that queues are only redeclared automatically when they are used by a listener annotated with @RabbitListener.
What I'm trying to figure out is how to achieve the same behavior for a manually created listener container. Just to clarify, I'm not talking about redeclaration on connection recovery — I mean when a queue or exchange is deleted while the connection remains active.

[artembilan]

The logic there in the AbstractMessageListenerContainer is like this:

	/**
	 * Use {@link AmqpAdmin#initialize()} to redeclare everything if necessary.
	 * Since auto deletion of a queue can cause upstream elements
	 * (bindings, exchanges) to be deleted too, everything needs to be redeclared if
	 * a queue is missing.
	 * Declaration is idempotent so, aside from some network chatter, there is no issue,
	 * and we only will do it if we detect our queue is gone.
	 * <p>
	 * In general, it makes sense only for the 'auto-delete' or 'expired' queues,
	 * but with the server TTL policy we don't have ability to determine 'expiration'
	 * option for the queue.
	 * <p>
	 * Starting with version 1.6, if
	 * {@link #setMismatchedQueuesFatal(boolean) mismatchedQueuesFatal} is true,
	 * the declarations are always attempted during restart so the listener will
	 * fail with a fatal error if mismatches occur.
	 */
	protected void redeclareElementsIfNecessary() {
		this.lifecycleLock.lock();
		try {
			AmqpAdmin admin = getAmqpAdmin();
			if (!this.lazyLoad && admin != null && isAutoDeclare()) {
				try {
					attemptDeclarations(admin);
					this.logDeclarationException.set(true);
				}

The isAutoDeclare() is true by default.
But AmqpAdmin has to be provided.
It is there in those containers which are beans and that instance is pulled from the ApplicationContext.

However this is still not enough.
That attemptDeclarations() is like this:

private void attemptDeclarations(AmqpAdmin admin) {
		ApplicationContext context = getApplicationContext();
		if (context != null) {
			Set<String> queueNames = getQueueNamesAsSet();

So, the ApplicationContext is also has to be there for those containers yo create manually.

Does that make sense now?

[tex1988]

It looks like that is what I need. Thank you for the explanation