Ensure reactive transaction rollback on commit error

This change fixes a situation where error handling was skipped during
`processCommit()` in case the `doCommit()` failed. The error handling
was set up via an `onErrorResume` operator that was nested inside a
`then(...)`, applied to an inner `Mono.empty()`. As a consequence,
it would never receive an error signal (effectively decoupling the
onErrorResume from the main chain).

This change simply moves the error handling back one level up. It also
simplifies the `doCommit` code a bit by getting rid of the steps that
artificially introduce a `Mono<Object>` return type, which is not really
needed.

A pre-existing test was missing the fact that the rollback didn't occur,
which is now fixed. Another dedicated test is introduced building upon
the `ReactiveTestTransactionManager` class.

Closes gh-30096

spring-tx/src/test/java/org/springframework/transaction/reactive/ReactiveTestTransactionManager.java

@@ -36,6 +36,8 @@ class ReactiveTestTransactionManager extends AbstractReactiveTransactionManager
 
 	private final boolean canCreateTransaction;
 
+	private final boolean forceFailOnCommit;
+
 	protected boolean begin = false;
 
 	protected boolean commit = false;
@@ -48,8 +50,13 @@ class ReactiveTestTransactionManager extends AbstractReactiveTransactionManager
 
 
 	ReactiveTestTransactionManager(boolean existingTransaction, boolean canCreateTransaction) {
+		this(existingTransaction, canCreateTransaction, false);
+	}
+
+	ReactiveTestTransactionManager(boolean existingTransaction, boolean canCreateTransaction, boolean forceFailOnCommit) {
 		this.existingTransaction = existingTransaction;
 		this.canCreateTransaction = canCreateTransaction;
+		this.forceFailOnCommit = forceFailOnCommit;
 	}
 
 
@@ -79,7 +86,12 @@ class ReactiveTestTransactionManager extends AbstractReactiveTransactionManager
 		if (!TRANSACTION.equals(status.getTransaction())) {
 			return Mono.error(new IllegalArgumentException("Not the same transaction object"));
 		}
-		return Mono.fromRunnable(() -> this.commit = true);
+		return Mono.fromRunnable(() -> {
+			this.commit = true;
+			if (this.forceFailOnCommit) {
+				throw new IllegalArgumentException("Forced failure on commit");
+			}
+		});
 	}
 
 	@Override

spring-tx/src/test/java/org/springframework/transaction/reactive/ReactiveTransactionSupportTests.java

@@ -203,6 +203,22 @@ public class ReactiveTransactionSupportTests {
 		assertHasCleanedUp(tm);
 	}
 
+	//gh-28968
+	@Test
+	void errorInCommitDoesInitiateRollbackAfterCommit() {
+		ReactiveTestTransactionManager tm = new ReactiveTestTransactionManager(false, true, true);
+		TransactionalOperator rxtx = TransactionalOperator.create(tm);
+
+		StepVerifier.create(rxtx.transactional(Mono.just("bar")))
+				.verifyErrorMessage("Forced failure on commit");
+
+		assertHasBegan(tm);
+		assertHasCommitted(tm);
+		assertHasRolledBack(tm);
+		assertHasNotSetRollbackOnly(tm);
+		assertHasCleanedUp(tm);
+	}
+
 	private void assertHasBegan(ReactiveTestTransactionManager actual) {
 		assertThat(actual.begin).as("Expected <ReactiveTransactionManager.begin()> but was <begin()> was not invoked").isTrue();
 	}