+ * If no custom translator is provided, a default {@link R2dbcExceptionSubclassTranslator} is used which translates + * {@link R2dbcException}'s subclasses into Springs {@link DataAccessException} hierarchy. + * + * @see R2dbcExceptionSubclassTranslator + * @since 1.1 + */ + public void setExceptionTranslator(R2dbcExceptionTranslator exceptionTranslator) { + this.exceptionTranslator = exceptionTranslator; + } + /** * Obtain the {@link ConnectionFactory} for actual use. * @@ -242,17 +261,22 @@ public class R2dbcTransactionManager extends AbstractReactiveTransactionManager } }).thenReturn(con).onErrorResume(e -> { - CannotCreateTransactionException ex = new CannotCreateTransactionException( - "Could not open R2DBC Connection for transaction", e); - if (txObject.isNewConnectionHolder()) { return ConnectionFactoryUtils.releaseConnection(con, obtainConnectionFactory()).doOnTerminate(() -> { txObject.setConnectionHolder(null, false); - }).then(Mono.error(ex)); + }).then(Mono.error(e)); } - return Mono.error(ex); + return Mono.error(e); }); + }).onErrorResume(e -> { + + CannotCreateTransactionException ex = new CannotCreateTransactionException( + "Could not open R2DBC Connection for transaction", + e instanceof R2dbcException ? potentiallyTranslateException("Open R2DBC Connection", (R2dbcException) e) + : e); + + return Mono.error(ex); }); }).then(); } @@ -321,7 +345,7 @@ public class R2dbcTransactionManager extends AbstractReactiveTransactionManager } return Mono.from(connection.commitTransaction()) - .onErrorMap(ex -> new TransactionSystemException("Could not commit R2DBC transaction", ex)); + .onErrorMap(R2dbcException.class, ex -> translateException("R2DBC commit", ex)); } /* @@ -339,7 +363,7 @@ public class R2dbcTransactionManager extends AbstractReactiveTransactionManager } return Mono.from(connection.rollbackTransaction()) - .onErrorMap(ex -> new TransactionSystemException("Could not roll back R2DBC transaction", ex)); + .onErrorMap(R2dbcException.class, ex -> translateException("R2DBC rollback", ex)); } /* @@ -496,6 +520,31 @@ public class R2dbcTransactionManager extends AbstractReactiveTransactionManager return null; } + /** + * Translate the given R2DBC commit/rollback exception to a common Spring exception to propagate from the + * {@link #commit}/{@link #rollback} call. + *
+ * The default implementation throws a {@link TransactionSystemException}. Subclasses may specifically identify + * concurrency failures etc. + * + * @param task the task description (commit or rollback). + * @param ex the SQLException thrown from commit/rollback. + * @return the translated exception to throw, either a {@link org.springframework.dao.DataAccessException} or a + * {@link org.springframework.transaction.TransactionException} + * @since 1.1 + */ + protected RuntimeException translateException(String task, R2dbcException ex) { + + Exception translated = potentiallyTranslateException(task, ex); + return new TransactionSystemException(task + " failed", translated); + } + + private Exception potentiallyTranslateException(String task, R2dbcException ex) { + + DataAccessException translated = exceptionTranslator.translate(task, null, ex); + return translated != null ? translated : ex; + } + /** * ConnectionFactory transaction object, representing a ConnectionHolder. Used as transaction object by * ConnectionFactoryTransactionManager. diff --git a/src/test/java/org/springframework/data/r2dbc/connectionfactory/R2dbcTransactionManagerUnitTests.java b/src/test/java/org/springframework/data/r2dbc/connectionfactory/R2dbcTransactionManagerUnitTests.java index 1d5d0c4e..ff45ef40 100644 --- a/src/test/java/org/springframework/data/r2dbc/connectionfactory/R2dbcTransactionManagerUnitTests.java +++ b/src/test/java/org/springframework/data/r2dbc/connectionfactory/R2dbcTransactionManagerUnitTests.java @@ -23,6 +23,7 @@ import static org.mockito.Mockito.*; import io.r2dbc.spi.Connection; import io.r2dbc.spi.ConnectionFactory; import io.r2dbc.spi.IsolationLevel; +import io.r2dbc.spi.R2dbcBadGrammarException; import io.r2dbc.spi.Statement; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; @@ -32,6 +33,8 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Before; import org.junit.Test; +import org.springframework.data.r2dbc.BadSqlGrammarException; +import org.springframework.transaction.CannotCreateTransactionException; import org.springframework.transaction.IllegalTransactionStateException; import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.reactive.TransactionSynchronization; @@ -94,6 +97,29 @@ public class R2dbcTransactionManagerUnitTests { assertThat(sync.afterCompletionCalled).isTrue(); } + @Test // gh-329 + public void testBeginFails() { + + reset(connectionFactoryMock); + when(connectionFactoryMock.create()).thenReturn(Mono.error(new R2dbcBadGrammarException("fail"))); + + when(connectionMock.rollbackTransaction()).thenReturn(Mono.empty()); + + DefaultTransactionDefinition definition = new DefaultTransactionDefinition(); + definition.setIsolationLevel(TransactionDefinition.ISOLATION_SERIALIZABLE); + + TransactionalOperator operator = TransactionalOperator.create(tm, definition); + + ConnectionFactoryUtils.getConnection(connectionFactoryMock).as(operator::transactional) // + .as(StepVerifier::create) // + .expectErrorSatisfies(actual -> { + + assertThat(actual).isInstanceOf(CannotCreateTransactionException.class) + .hasCauseInstanceOf(BadSqlGrammarException.class); + + }).verify(); + } + @Test // gh-107 public void appliesIsolationLevel() { @@ -214,9 +240,11 @@ public class R2dbcTransactionManagerUnitTests { public void testCommitFails() { when(connectionMock.commitTransaction()).thenReturn(Mono.defer(() -> { - return Mono.error(new IllegalStateException("Commit should fail")); + return Mono.error(new R2dbcBadGrammarException("Commit should fail")); })); + when(connectionMock.rollbackTransaction()).thenReturn(Mono.empty()); + TransactionalOperator operator = TransactionalOperator.create(tm); ConnectionFactoryUtils.getConnection(connectionFactoryMock) // @@ -225,7 +253,7 @@ public class R2dbcTransactionManagerUnitTests { }).then() // .as(operator::transactional) // .as(StepVerifier::create) // - .verifyError(); + .verifyError(IllegalTransactionStateException.class); verify(connectionMock).isAutoCommit(); verify(connectionMock).beginTransaction(); @@ -263,6 +291,35 @@ public class R2dbcTransactionManagerUnitTests { verifyNoMoreInteractions(connectionMock); } + @Test // gh-329 + public void testRollbackFails() { + + when(connectionMock.rollbackTransaction()).thenReturn(Mono.defer(() -> { + return Mono.error(new R2dbcBadGrammarException("Commit should fail")); + }), Mono.empty()); + + TransactionalOperator operator = TransactionalOperator.create(tm); + + operator.execute(reactiveTransaction -> { + + reactiveTransaction.setRollbackOnly(); + + return ConnectionFactoryUtils.getConnection(connectionFactoryMock) // + .doOnNext(it -> { + it.createStatement("foo"); + }).then(); + }).as(StepVerifier::create) // + .verifyError(IllegalTransactionStateException.class); + + verify(connectionMock).isAutoCommit(); + verify(connectionMock).beginTransaction(); + verify(connectionMock).createStatement("foo"); + verify(connectionMock, never()).commitTransaction(); + verify(connectionMock).rollbackTransaction(); + verify(connectionMock).close(); + verifyNoMoreInteractions(connectionMock); + } + @Test // gh-107 public void testTransactionSetRollbackOnly() {