Unable to use ConnectionFactoryUtils.checkAfterCompletion() properly

[tbadie]

Hi everyone,

In our spring application we are using a transaction to mysql and to rabbitmq. We are using spring boot 3.2.4 and spring amqp 3.1.3.

We rely on the Best Effort One Phase Commit, with a commit to the DB and then Rabbitmq committing. We wanted to make use of the checkAfterCompletion method described here: https://docs.spring.io/spring-amqp/reference/amqp/transactions.html#tx-sync to log when the Rabbitmq transaction fails but the DB one succeeded.

However we're not quite sure where to set that up. We have tried to create a non-transactional receiver which delegates to a transactional one:

public class MessageProcessorWrapper {
     ...

    @RabbitListener(queues = "${messaging.queues.main}")
    public void receiveMessage(Message message) {
        // Aware that this is a global config and doesn't need to be set every time...
        // But for the example sake we're showing it here.
        ConnectionFactoryUtils.enableAfterCompletionFailureCapture(true);

        messageProcessor.receiveMessage(message);

        try {
            ConnectionFactoryUtils.checkAfterCompletion();
        } catch (AfterCompletionFailedException exception) {
            LOGGER.error("The rabbit transaction failed!", exception);
        }
    }
}

With the message processor looking like:

public class MessageProcessor {
    @Transactional
    public void receiveMessage(Message message) {
       ....
    }
}

The chain of event as I understand it is:

1. org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer#executeListener seems to start the flow (so presumably a rabbit transaction is started?)
2. Our rabbit listener method above is called
3. We call our messageProcessor and a transaction is started
4. The messageProcessor returns and the DB transaction is committed
5. checkAfterCompletion is called but there is nothing to check since the rabbitmq transaction hasn't yet been committed
   org.springframework.amqp.rabbit.connection.RabbitUtils#commitIfNecessary is eventually called and may fail without us being able to be notified by it.

There is no obvious place where I can call the checkAfterCompletion after that rabbitmq commit occurs and google doesn't seem to have examples.

Do you have any advice on how to move forward with this please? I hope this is the right place to ask the question.

[artembilan]

The transaction is started around @RabbitListener POJO method call only if AbstractMessageListenerContainer is populated with a PlatformTransactionManager.
The logic is like this:

	private boolean receiveAndExecute(final BlockingQueueConsumer consumer) throws Exception { // NOSONAR

		PlatformTransactionManager transactionManager = getTransactionManager();
		if (transactionManager != null) {
			try {
				if (this.transactionTemplate == null) {
					this.transactionTemplate =
							new TransactionTemplate(transactionManager, getTransactionAttribute());
				}
				return this.transactionTemplate
						.execute(status -> { // NOSONAR null never returned
							RabbitResourceHolder resourceHolder = ConnectionFactoryUtils.bindResourceToTransaction(
									new RabbitResourceHolder(consumer.getChannel(), false),
									getConnectionFactory(), true);
							// unbound in ResourceHolderSynchronization.beforeCompletion()
							try {
								return doReceiveAndExecute(consumer);
							}
							catch (RuntimeException e1) {
								prepareHolderForRollback(resourceHolder, e1);
								throw e1;
							}
							catch (Exception e2) {
								throw new WrappedTransactionException(e2);
							}
						});
			}
			catch (WrappedTransactionException e) { // NOSONAR exception flow control
				throw (Exception) e.getCause();
			}
		}

		return doReceiveAndExecute(consumer);

	}

So, when Rabbit transaction is there, it is indeed wrong to use ConnectionFactoryUtils.checkAfterCompletion(); inside a @RabbitListener method since transaction is going to complete when we return control from that method back to the container.

Looks like the ConnectionFactoryUtils.checkAfterCompletion(); was designed for RabbitTemplate since this is only what I can find in tests:

		RabbitTemplate template = new RabbitTemplate(connectionFactory);
		template.setChannelTransacted(true);
		TransactionTemplate tt = new TransactionTemplate(new TestTransactionManager());
		tt.execute(status -> {
			template.convertAndSend("foo", "bar");
			return null;
		});
		assertThat(TransactionSynchronizationManager.hasResource(connectionFactory)).isFalse();
		assertThatIllegalStateException()
			.isThrownBy(() -> (TransactionSynchronizationManager.getSynchronizations()).isEmpty())
			.withMessage("Transaction synchronization is not active");
		assertThatExceptionOfType(AfterCompletionFailedException.class)
			.isThrownBy(() -> ConnectionFactoryUtils.checkAfterCompletion());

I think the mentioned logic in the SimpleMessageListenerContainer could be improved to call that ConnectionFactoryUtils.checkAfterCompletion() in the finally block of that transactionTemplate.execute().

As a workaround I only see a way to set TransactionInterceptor and some other custom MethodInterceptor with a ConnectionFactoryUtils.checkAfterCompletion() call for the:

	/**
	 * Public setter for the {@link Advice} to apply to listener executions.
	 * <p>
	 * If a {code #setTransactionManager(PlatformTransactionManager) transactionManager} is provided as well, then
	 * separate advice is created for the transaction and applied first in the chain. In that case the advice chain
	 * provided here should not contain a transaction interceptor (otherwise two transactions would be be applied).
	 * @param adviceChain the advice chain to set
	 */
	public void setAdviceChain(Advice... adviceChain) {

Does that make sense to you?

[tbadie]

Hi @artembilan,

Thanks again for your answer. I indeed see that mistake and I should have spotted that based on the javadoc of the adviceChain method. Sorry about that.

Unfortunately, if I remove the transactionManager, the rabbitmq commit happens in org.springframework.amqp.rabbit.connection.RabbitUtils#commitIfNecessary which is after the advices are executed and I'm back to the same place of checking whether there was an issue before any issue could occur. So I feel that maybe this workaround is just not meant to work for me.

Given that I can't keep bothering you with question until this workaround works, I think it's probably time to try to make a contribution instead. That way more people can benefit from it.

Do we agree that the code change that you meant is this?

    private boolean receiveAndExecute(final BlockingQueueConsumer consumer) throws Exception { // NOSONAR

        PlatformTransactionManager transactionManager = getTransactionManager();
        if (transactionManager != null) {
            try {
                if (this.transactionTemplate == null) {
                    this.transactionTemplate =
                            new TransactionTemplate(transactionManager, getTransactionAttribute());
                }
                return this.transactionTemplate
                        .execute(status -> { // NOSONAR null never returned
                            RabbitResourceHolder resourceHolder = ConnectionFactoryUtils.bindResourceToTransaction(
                                    new RabbitResourceHolder(consumer.getChannel(), false),
                                    getConnectionFactory(), true);
                            // unbound in ResourceHolderSynchronization.beforeCompletion()
                            try {
                                return doReceiveAndExecute(consumer);
                            }
                            catch (RuntimeException e1) {
                                prepareHolderForRollback(resourceHolder, e1);
                                throw e1;
                            }
                            catch (Exception e2) {
                                throw new AbstractMessageListenerContainer.WrappedTransactionException(e2);
                            }
                        });
            }
            catch (AbstractMessageListenerContainer.WrappedTransactionException e) { // NOSONAR exception flow control
                throw (Exception) e.getCause();
            }
            finally {
                ConnectionFactoryUtils.checkAfterCompletion();
            }
        }

        return doReceiveAndExecute(consumer);

    }

That way:

1. The check occurs outside of the transaction
2. The exception gets a chance to bubble up and get caught appropriately.

So all I'd need to do is add a unit test as well. Did I get that right?

[artembilan]

Well, let's look if that simpleRabbitListenerContainerFactory.setChannelTransacted(Boolean.TRUE); is a cause!
You start transaction yourself in the advice, so no any local transaction for the container then!

Yes, I think your plan sounds good about contribution.

Thank you!

[tbadie]

I've removed that line and actually it worked. I couldn't quite believe it so I dug a bit more into it: it worked because the listener I was using broadcast another event to a transactional channel, which meant that this piece of code was called: org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils#getTransactionalResourceHolder(org.springframework.amqp.rabbit.connection.ConnectionFactory, boolean, boolean) and in turn meant that the JpaTransactionManager was now aware of the rabbit synchronization. That's what allowed the proper commit of both transactions (JPA and Rabbit) before reaching my CheckAfterCompletionInterceptor.

So I would have to call this myself within the transaction, but I'm worried about going too much off road with the expectation and making future migrations a nightmare. So I'll give up on the workaround, but thanks for guiding me through the complexity of Spring AMQP and transactions!

I still have one question regarding the code for the contribution. I'd like to actually be able to log some information about the event that has failed. But based on the code that seems impossible: at the level where I'd call the ConnectionFactoryUtils.checkAfterCompletion() method I would only have access to the BlockingQueueConsumer consumer and the message is only available in the transaction itself here and it's not a repeatable call. Could you please confirm that?

Thanks!

[artembilan]

Well, unfortunately that's what you can put into an exception your throw from your listener method.
This ConnectionFactoryUtils.checkAfterCompletion() is exactly for transaction, so we cannot do anything with the message since we have to come back from the consumer to commit it in the RabbitResourceHolder.commitAll().

[artembilan]

The mentioned issue is here: #2805