TransactionalOperator should not attempt to rollback after a failed commit

[EnricSala]

Expected behavior

Using reactive transaction management, when a commit step fails in the TransactionalOperator, then the commit failure exception should be propagated. This can be for example a ConcurrencyFailureException.

Actual behavior (Spring 5.3.9)

When a commit fails, the commit exception is logged & dropped in TransactionalOperatorImpl, and instead an IllegalTransactionStateException is propagated.

Practical case

When under high concurrency, our application is observing uninformative IllegalTransactionStateException like this:

org.springframework.transaction.IllegalTransactionStateException: Transaction is already completed - do not call commit or rollback more than once per transaction
	at org.springframework.transaction.reactive.AbstractReactiveTransactionManager.rollback(AbstractReactiveTransactionManager.java:492)
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Assembly trace from producer [reactor.core.publisher.MonoLift] :
	reactor.core.publisher.Mono.error
	org.springframework.transaction.reactive.AbstractReactiveTransactionManager.rollback(AbstractReactiveTransactionManager.java:492)
Error has been observed at the following site(s):
	|_          Mono.error ⇢ at org.springframework.transaction.reactive.AbstractReactiveTransactionManager.rollback(AbstractReactiveTransactionManager.java:492)
	|_     Mono.onErrorMap ⇢ at org.springframework.transaction.reactive.TransactionalOperatorImpl.rollbackOnException(TransactionalOperatorImpl.java:119)
	|_           Mono.then ⇢ at org.springframework.transaction.reactive.TransactionalOperatorImpl.lambda$null$2(TransactionalOperatorImpl.java:83)
	|_  Mono.onErrorResume ⇢ at org.springframework.transaction.reactive.TransactionalOperatorImpl.lambda$null$3(TransactionalOperatorImpl.java:83)
	|_        Mono.flatMap ⇢ at org.springframework.transaction.reactive.TransactionalOperatorImpl.lambda$transactional$4(TransactionalOperatorImpl.java:81)
	|_        Mono.flatMap ⇢ at org.springframework.transaction.reactive.TransactionalOperatorImpl.transactional(TransactionalOperatorImpl.java:75)
	|_   Mono.contextWrite ⇢ at org.springframework.transaction.reactive.TransactionalOperatorImpl.transactional(TransactionalOperatorImpl.java:85)
	|_   Mono.contextWrite ⇢ at org.springframework.transaction.reactive.TransactionalOperatorImpl.transactional(TransactionalOperatorImpl.java:86)

The above illegal transaction state seems to be overriding the propagation of this exception:

ERROR --- o.s.t.r.TransactionalOperatorImpl : Application exception overridden by rollback exception

java.lang.RuntimeException: Async resource cleanup failed after onComplete
	at reactor.core.publisher.FluxUsingWhen$CommitInner.onError(FluxUsingWhen.java:533)
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Assembly trace from producer [reactor.core.publisher.MonoLift] :
	reactor.core.publisher.Mono.usingWhen
	org.springframework.transaction.reactive.TransactionalOperatorImpl.lambda$null$3(TransactionalOperatorImpl.java:81)
Error has been observed at the following site(s):
	|_ Mono.usingWhen ⇢ at org.springframework.transaction.reactive.TransactionalOperatorImpl.lambda$null$3(TransactionalOperatorImpl.java:81)
Stack trace:
		at reactor.core.publisher.FluxUsingWhen$CommitInner.onError(FluxUsingWhen.java:533)
		...
		at io.r2dbc.postgresql.util.FluxDiscardOnCancel$FluxDiscardOnCancelSubscriber.onComplete(FluxDiscardOnCancel.java:99)
		at io.r2dbc.postgresql.client.ReactorNettyClient$Conversation.complete(ReactorNettyClient.java:719)
		at io.r2dbc.postgresql.client.ReactorNettyClient$BackendMessageSubscriber.emit(ReactorNettyClient.java:984)
		...
Caused by: org.springframework.dao.ConcurrencyFailureException: R2DBC commit; could not serialize access due to read/write dependencies among transactions;
	at org.springframework.r2dbc.connection.ConnectionFactoryUtils.convertR2dbcException(ConnectionFactoryUtils.java:218)
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Assembly trace from producer [reactor.core.publisher.MonoLift] :
	reactor.core.publisher.Mono.onErrorMap
	org.springframework.r2dbc.connection.R2dbcTransactionManager.doCommit(R2dbcTransactionManager.java:279)
Error has been observed at the following site(s):
	|_    Mono.onErrorMap ⇢ at org.springframework.r2dbc.connection.R2dbcTransactionManager.doCommit(R2dbcTransactionManager.java:279)
	|_                    ⇢ at org.springframework.transaction.reactive.AbstractReactiveTransactionManager.lambda$processCommit$21(AbstractReactiveTransactionManager.java:445)
	|_         Mono.defer ⇢ at org.springframework.transaction.reactive.AbstractReactiveTransactionManager.processCommit(AbstractReactiveTransactionManager.java:439)
	|_          Mono.then ⇢ at org.springframework.transaction.reactive.AbstractReactiveTransactionManager.processCommit(AbstractReactiveTransactionManager.java:439)
	|_          Mono.then ⇢ at org.springframework.transaction.reactive.AbstractReactiveTransactionManager.processCommit(AbstractReactiveTransactionManager.java:448)
	|_          Mono.then ⇢ at org.springframework.transaction.reactive.AbstractReactiveTransactionManager.processCommit(AbstractReactiveTransactionManager.java:474)
	|_         Mono.error ⇢ at org.springframework.transaction.reactive.AbstractReactiveTransactionManager.lambda$processCommit$25(AbstractReactiveTransactionManager.java:480)
	|_          Mono.then ⇢ at org.springframework.transaction.reactive.AbstractReactiveTransactionManager.lambda$processCommit$25(AbstractReactiveTransactionManager.java:480)
	|_ Mono.onErrorResume ⇢ at org.springframework.transaction.reactive.AbstractReactiveTransactionManager.processCommit(AbstractReactiveTransactionManager.java:479)
	|_          Mono.then ⇢ at org.springframework.transaction.reactive.AbstractReactiveTransactionManager.processCommit(AbstractReactiveTransactionManager.java:480)
	|_                    ⇢ at org.springframework.transaction.reactive.AbstractReactiveTransactionManager.lambda$commit$20(AbstractReactiveTransactionManager.java:420)
	|_       Mono.flatMap ⇢ at org.springframework.transaction.reactive.AbstractReactiveTransactionManager.commit(AbstractReactiveTransactionManager.java:412)
Stack trace:
		at org.springframework.r2dbc.connection.ConnectionFactoryUtils.convertR2dbcException(ConnectionFactoryUtils.java:218)
		at org.springframework.r2dbc.connection.R2dbcTransactionManager.translateException(R2dbcTransactionManager.java:439)
		at org.springframework.r2dbc.connection.R2dbcTransactionManager.lambda$doCommit$9(R2dbcTransactionManager.java:279)
		...

Investigation of the cause

The problem described above matches this warning on ReactiveTransactionManager#rollback:

spring-framework/spring-tx/src/main/java/org/springframework/transaction/ReactiveTransactionManager.java

Lines 99 to 102 in fb7eea9

	* <p><b>Do not call rollback on a transaction if commit threw an exception.</b>
	* The transaction will already have been completed and cleaned up when commit
	* returns, even in case of a commit exception. Consequently, a rollback call
	* after commit failure will lead to an IllegalTransactionStateException.

Looking at the current implementation of TransactionalOperatorImpl, it looks like indeed this can happen; if transactionManager::commit emits an error then a rollback will be attempted by the onErrorResume:

spring-framework/spring-tx/src/main/java/org/springframework/transaction/reactive/TransactionalOperatorImpl.java

Lines 74 to 84 in fb7eea9

	public <T> Mono<T> transactional(Mono<T> mono) {
	return TransactionContextManager.currentContext().flatMap(context -> {
	Mono<ReactiveTransaction> status = this.transactionManager.getReactiveTransaction(this.transactionDefinition);
	// This is an around advice: Invoke the next interceptor in the chain.
	// This will normally result in a target object being invoked.
	// Need re-wrapping of ReactiveTransaction until we get hold of the exception
	// through usingWhen.
	return status.flatMap(it -> Mono.usingWhen(Mono.just(it), ignore -> mono,
	this.transactionManager::commit, (res, err) -> Mono.empty(), this.transactionManager::rollback)
	.onErrorResume(ex -> rollbackOnException(it, ex).then(Mono.error(ex))));
	})

Looks related to the discussion in #23562

Possible solution

The operators in TransactionalOperatorImpl#transactional could be restructured in such a way that rollbacks are not attempted after a failure to commit. Maybe this could be accomplished by moving the rollbackOnException to the asyncError parameter in the usingWhen. A similar approach should be applied in TransactionalOperatorImpl#execute.

Additionally, would it make sense to add the application exception as suppressed when the rollback in rollbackOnException emits an exception? This could make it easier to debug this and similar issues because currently the application exception is only linked when the rollback emits an TransactionSystemException, but not in other cases like IllegalTransactionStateException.

[jhoeller]

@mp911de any advice on this?

[mp911de]

The main difference between the imperative and the reactive transaction handling is that TransactionTemplate handles exceptions first and then falls back to commit when no exception has happened. TransactionalOperatorImpl handles the commit before rollbackOnException.

Reactor's usingWhen suggests this type of design. Moving onErrorResume to inside the usingWhen action closure should address the issue. asyncError doesn't consider errors from asyncComplete so moving error handling to inside of resourceClosure looks like an appropriate fix.

[AntonioMorales97]

What is the status on this? This is a very standard use case and our current solution is basically a copy-paste of the TransactionalOperatorImpl where we avoid rollbacks for ConcurrencyFailureException.

[sdeleuze]

I have asked a feedback on the latest version of the PR here and will have a look after Mark feedback.

[mp911de]

This issue is related to #28968, I'll have another look as this PR contains already fixes that I started preparing for #28968

[AntonioMorales97]

We had to implement our custom reactive transaction manager for this but when we upgraded to Spring Boot 3 and updated r2dbc it stopped working. For some reason the isolation level cannot be set/passed to r2dbc. This can be reproduced with the following test (the second exception is thrown):

@Transactional(isolation = Isolation.SERIALIZABLE)
public Mono<Void> runAndEnsureSerializable() {
    return Mono.deferContextual(ctx -> {
        var transactionContext = ctx.get(TransactionContext.class);

        if (transactionContext.getCurrentTransactionIsolationLevel() != Isolation.SERIALIZABLE.value()) {
            throw new RuntimeException("Expected isolation level to be SERIALIZABLE");
        }

        return db.inConnection(connection -> {
            if (connection.getTransactionIsolationLevel() != IsolationLevel.SERIALIZABLE) {
                throw new RuntimeException("Expected isolation level to be SERIALIZABLE");
            }

            return Mono.empty();
        });
    });
}

and the following configuration:

@Bean
public ConnectionFactory connectionFactory(PostgreSQLContainer postgres) {
    return new ConnectionPool(ConnectionPoolConfiguration.builder()
            .initialSize(0)
            .validationDepth(ValidationDepth.REMOTE)
            .connectionFactory(ConnectionFactories.get(ConnectionFactoryOptions.builder()
                    .option(DRIVER, "postgresql")
                    .option(HOST, postgres.getHost())
                    .option(PORT, postgres.getFirstMappedPort())
                    .option(USER, postgres.getUsername())
                    .option(PASSWORD, postgres.getPassword())
                    .option(DATABASE, postgres.getDatabaseName())
                    .build()))
            .customizer(builder -> builder.releaseHandler(Connection::rollbackTransaction))
            .build());
}

We are using:

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>3.0.2</version>
</parent>
...
<dependency>
   <groupId>io.r2dbc</groupId>
   <artifactId>r2dbc-spi</artifactId>
   <version>1.0.0.RELEASE</version>
</dependency>
<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>r2dbc-postgresql</artifactId>
    <version>1.0.0.RELEASE</version>
</dependency>
<dependency>
    <groupId>io.r2dbc</groupId>
    <artifactId>r2dbc-pool</artifactId>
</dependency>

The problem seems to be that autocommit is switched in spring-r2dbc:6.0.4 but not in spring-r2dbc:5.3.24:

spring-r2dbc:6.0.4:

connectionMono.flatMap(con -> switchAutoCommitIfNecessary(con, transaction)
      .then(Mono.from(doBegin(definition, con)))
      .then(prepareTransactionalConnection(con, definition))

spring-r2dbc:5.3.24:

connectionMono.flatMap(con -> prepareTransactionalConnection(con, definition, transaction)
      .then(Mono.from(con.beginTransaction()))
      .then(prepareTransactionalConnection(con, definition))

Any ideas?

[sdeleuze]

@AntonioMorales97 Please create a dedicated issue if that's another bug.

This issue will be handle via #27572 so I close it as a duplicate.