Send of message via tx-channel is commited early when external (JTA) transaction is present [AMQP-479]

[spring-operator]

Norbert Wirges opened AMQP-479 and commented

When an external JTA/EJB transaction calls a transactional spring bean (using JtaTransactionManager), which in turn sends a RabbitMQ-message (using RabbitTemplate), then the message is committed on the channel when the spring transaction bracket ends (not as expected when the outer JTA transaction is committed).

This StackOverflow question describes the same problem:
http://stackoverflow.com/questions/25773000/spring-amqp-rabbitmq-transaction-rollback-in-ejb3-cmt

The reason seems to be, that AbstractPlatformTransactionManager.processCommit() executes onCommit() callbacks immediately (whereas afterCompletion-Callbacks are passed on to a surrounding transaction). As the RabbitTemplate uses onCommit() callbacks (see RabbitResourceSynchronization) this causes the described behavior.
Moving the channel commit to afterCompletion in RabbitResourceSynchronization should fix this problem.

---

Affects: 1.3.8, 1.4.3

Reference URL: http://stackoverflow.com/questions/25773000/spring-amqp-rabbitmq-transaction-rollback-in-ejb3-cmt

Referenced from: pull request #340

[spring-operator]

Artem Bilan commented

Norbert Wirges, I haven't looked to the solution yet, but feel free to sign CLA (https://github.com/spring-projects/spring-integration/blob/master/CONTRIBUTING.md) and PullRequest for the fix.

[spring-operator]

Artem Bilan commented

Norbert Wirges, sorry for delay.

Looking to your SO question you are using CMT and Spring JTA support (@Transactional) for you underlying service.
Would you mind trying the same but with BMT, not CMT?

My concern that JTA (UserTransaction) doesn't see CMT context to join, hence your main rollback doesn't effect inner JTA transaction.

The thought comes from the fact that RabbitResourceSynchronization is fully similar to the JmsResourceSynchronization. That's why I'm not sure that we are liable to change anything here.

Please, let us know how it works with BMT.

Thank you!

[spring-operator]

Norbert Wirges commented

Actually the SO question isn't mine - I'm just a guy with the same problem.

I guess JmsResourceHolder works due to the caught TransactionInProgressException in JmsResourceHolder.commitAll() (JMS is typically managed by the JEE server and doesn't let you commit manually when a JTA transaction is running).

How we handle the transactions (CMT/BMT) in the JEE part doesn't seem to matter - either way Spring "sees" the JTA transaction fine (DefaultTransactionStatus.isNewTransaction() is false).

This is a stack excerpt from the moment the 'early' commit happens (here with CMT):

Thread [WorkerThread#0[172.20.111.20:49288]] (Suspended (breakpoint at line 151 in RabbitResourceHolder))
RabbitResourceHolder.commitAll() line: 151
ConnectionFactoryUtils$RabbitResourceSynchronization.processResourceAfterCommit(RabbitResourceHolder) line: 255
ConnectionFactoryUtils$RabbitResourceSynchronization.processResourceAfterCommit(ResourceHolder) line: 235
ConnectionFactoryUtils$RabbitResourceSynchronization(ResourceHolderSynchronization).afterCommit() line: 85
TransactionSynchronizationUtils.invokeAfterCommit(List<TransactionSynchronization>) line: 133
TransactionSynchronizationUtils.triggerAfterCommit() line: 121
JtaTransactionManager(AbstractPlatformTransactionManager).triggerAfterCommit(DefaultTransactionStatus) line: 954
JtaTransactionManager(AbstractPlatformTransactionManager).processCommit(DefaultTransactionStatus) line: 799
JtaTransactionManager(AbstractPlatformTransactionManager).commit(TransactionStatus) line: 726
TransactionTemplate.execute(TransactionCallback<T>) line: 150
SpringDelegatorImpl.delegateToSpringBean(SpringDelegationCall) line: 101 [project helper class for EJB -> Spring calls]
...
TxInterceptorCMT(AbstractTxInterceptor).invokeNext(Invocation, boolean) line: 121
TxInterceptorCMT.runWithTransactions(Invocation) line: 378
TxInterceptorCMT.invoke(Invocation) line: 181
...

As you can see, the early commit is caused by AbstractPlatformTransactionManager.triggerAfterCommit().
AbstractPlatformTransactionManager.doCommit() is not called before (which is correct, as the 'real' commit is handled by the outer CMT/BMT).
For the JTA/RabbitMQ combination to work cleanly, the channel commit would need to be handled in AbstractPlatformTransactionManager.triggerAfterCompletion() (see JtaTransactionManager.doRegisterAfterCompletionWithJtaTransaction(), which moves Synchronizations to the outer JTA transaction)

[spring-operator]

Artem Bilan commented

First of all sorry for delay.
From one side we were busy with other stuff around.
From other side I might be smarter a bit this time :).

I see your point regarding JmsResourceSynchronization and it doesn't call :

@Override
protected void processResourceAfterCommit(JmsResourceHolder resourceHolder) {
	try {
		resourceHolder.commitAll();
	}
	catch (JMSException ex) {
		throw new SynchedLocalTransactionFailedException("Local JMS transaction failed to commit", ex);
	}
}

if it is transacted:

public void afterCommit() {
	if (!shouldReleaseBeforeCompletion()) {
		processResourceAfterCommit(this.resourceHolder);
	}
}

Since there is no any afterCompletion() impl in the JmsResourceSynchronization, I guess the real commit is done by XA tx on its registered resources, e.g. ActiveMQXASession.
The transacted fact depends on:

    /**
 * Set the transaction mode that is used when creating a JMS {@link Session}.
 * Default is "false".
 * <p>Note that within a JTA transaction, the parameters passed to
 * {@code create(Queue/Topic)Session(boolean transacted, int acknowledgeMode)}
 * method are not taken into account. Depending on the Java EE transaction context,
 * the container makes its own decisions on these values. Analogously, these
 * parameters are not taken into account within a locally managed transaction
 * either, since the accessor operates on an existing JMS Session in this case.
 * <p>Setting this flag to "true" will use a short local JMS transaction
 * when running outside of a managed transaction, and a synchronized local
 * JMS transaction in case of a managed transaction (other than an XA
 * transaction) being present. This has the effect of a local JMS
 * transaction being managed alongside the main transaction (which might
 * be a native JDBC transaction), with the JMS transaction committing
 * right after the main transaction.
 * @see javax.jms.Connection#createSession(boolean, int)
 */
public void setSessionTransacted(boolean sessionTransacted) {
     this.sessionTransacted = sessionTransacted;
   }

With RabbitMQ we don't have such a hook, because it doesn't support XA: http://lists.rabbitmq.com/pipermail/rabbitmq-discuss/2008-November/001931.html.

But anyway I think we can find some hook to trigger resourceHolder.commitAll(); in the afterCompletion() in case of an external Transaction:

  /**
 * Return whether the present transaction is new (else participating
 * in an existing transaction, or potentially not running in an
 * actual transaction in the first place).
 */
boolean isNewTransaction();

[spring-operator]

Norbert Wirges commented

Thanks - I was starting to think you guys had given up on this ;)

I still think the solution is really simple - in RabbitResourceSynchronization just move the resourceHolder.commitAll(); to afterCompletion():

    protected void processResourceAfterCommit(RabbitResourceHolder resourceHolder) {
       //resourceHolder.commitAll();
    }

    public void afterCompletion(int status) {
        if (status == TransactionSynchronization.STATUS_COMMITTED) {
            resourceHolder.commitAll();
        }
       ...
    }

This will work for non JTA as well as JTA contexts. The only behaviour change this would cause is the propagation of exceptions: exceptions in afterCommit() are propagated to the caller, while afterCompletion-exceptions are only logged.

[spring-operator]

Artem Bilan commented

Norbert Wirges,

thank you for feedback!

Please, refer to my Pull Request on the matter #340 for further discussion.

Yes, I admit that afterCompletion() is always invoked, but we really can't do that unconditionally. A but dangerous for existing application to upgrade.

[spring-operator]

Gary Russell commented

1.5.2 is now available.