Skip to content

Isolate transaction execution listener failures #37616

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fudianchn wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Isolate transaction execution listener failures #37616

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
02d693c
Add draft tests for transaction listener failures
fudianchn Oct 31, 2026
8e18152
Merge upstream main for transaction listener tests
fudianchn Oct 31, 2026
6f86c4a
Isolate transaction execution listener failures
fudianchn Oct 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
202 changes: 202 additions & 0 deletions ...ava/org/springframework/jdbc/datasource/TransactionExecutionListenerIntegrationTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,202 @@
/*
* Copyright 2002-present the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.springframework.jdbc.datasource;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;

import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionExecution;
import org.springframework.transaction.TransactionExecutionListener;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.transaction.support.TransactionTemplate;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatRuntimeException;
import static org.mockito.Mockito.mock;

class TransactionExecutionListenerIntegrationTests {

private final EmbeddedDatabase database = new EmbeddedDatabaseBuilder()
.generateUniqueName(true).setType(EmbeddedDatabaseType.H2).build();

private final List<Connection> connections = new ArrayList<>();

private final DelegatingDataSource dataSource = new DelegatingDataSource(this.database) {
@Override
public Connection getConnection() throws SQLException {
Connection connection = super.getConnection();
connections.add(connection);
return connection;
}
};

private final JdbcTemplate jdbc = new JdbcTemplate(this.dataSource);

private final DataSourceTransactionManager manager = new DataSourceTransactionManager(this.dataSource);

private final TransactionTemplate transaction = new TransactionTemplate(this.manager);


@BeforeEach
void createTable() {
this.jdbc.execute("create table entries (id integer)");
}

@AfterEach
void verifyCleanup() throws SQLException {
try {
assertThat(TransactionSynchronizationManager.getResourceMap()).isEmpty();
assertThat(TransactionSynchronizationManager.isSynchronizationActive()).isFalse();
assertThat(TransactionSynchronizationManager.isActualTransactionActive()).isFalse();
for (Connection connection : this.connections) {
assertThat(connection.isClosed()).isTrue();
}
}
finally {
TransactionSynchronizationManager.unbindResourceIfPossible(this.dataSource);
TransactionSynchronizationManager.clear();
for (Connection connection : this.connections) {
if (!connection.isClosed()) {
connection.rollback();
connection.close();
}
}
this.database.shutdown();
}
}

@ParameterizedTest
@ValueSource(strings = {"beforeBegin", "afterBegin", "beforeCommit", "afterCommit"})
void commitWithFailingListener(String callback) {
this.manager.addListener(failingListener(callback, false));

this.transaction.executeWithoutResult(status -> this.jdbc.update("insert into entries values (1)"));

assertThat(this.jdbc.queryForList("select id from entries", Integer.class)).containsExactly(1);
}

@ParameterizedTest
@ValueSource(booleans = {false, true})
void rollbackWithFailingListenerDoesNotCommitChanges(boolean rollbackOnly) {
this.manager.addListener(failingListener("beforeRollback", false));
RuntimeException businessFailure = new IllegalStateException("Business failure");

if (rollbackOnly) {
this.transaction.executeWithoutResult(status -> {
this.jdbc.update("insert into entries values (1)");
status.setRollbackOnly();
});
}
else {
assertThatRuntimeException().isThrownBy(() -> this.transaction.executeWithoutResult(status -> {
this.jdbc.update("insert into entries values (1)");
throw businessFailure;
})).isSameAs(businessFailure);
}

assertThat(this.jdbc.queryForList("select id from entries", Integer.class)).isEmpty();
}

@ParameterizedTest
@ValueSource(strings = {"beforeBegin", "afterBegin", "beforeRollback", "afterRollback"})
void nestedRollbackWithFailingListenerPreservesOuterTransaction(String callback) {
this.manager.addListener(failingListener(callback, true));
TransactionTemplate nested = nestedTransaction(TransactionDefinition.PROPAGATION_NESTED);
RuntimeException businessFailure = new IllegalStateException("Business failure");

this.transaction.executeWithoutResult(outer -> {
Object resource = TransactionSynchronizationManager.getResource(this.dataSource);
this.jdbc.update("insert into entries values (1)");
assertThatRuntimeException().isThrownBy(() -> nested.executeWithoutResult(inner -> {
this.jdbc.update("insert into entries values (2)");
throw businessFailure;
})).isSameAs(businessFailure);
assertThat(TransactionSynchronizationManager.getResource(this.dataSource)).isSameAs(resource);
assertThat(TransactionSynchronizationManager.isActualTransactionActive()).isTrue();
this.jdbc.update("insert into entries values (3)");
});

assertThat(this.jdbc.queryForList("select id from entries order by id", Integer.class)).containsExactly(1, 3);
}

@ParameterizedTest
@ValueSource(strings = {"beforeCommit", "afterCommit"})
void nestedCommitWithFailingListenerPreservesOuterTransaction(String callback) {
this.manager.addListener(failingListener(callback, true));
TransactionTemplate nested = nestedTransaction(TransactionDefinition.PROPAGATION_NESTED);

this.transaction.executeWithoutResult(outer -> {
this.jdbc.update("insert into entries values (1)");
nested.executeWithoutResult(inner -> this.jdbc.update("insert into entries values (2)"));
});

assertThat(this.jdbc.queryForList("select id from entries order by id", Integer.class)).containsExactly(1, 2);
}

@ParameterizedTest
@ValueSource(strings = {"beforeBegin", "afterBegin", "beforeCommit", "afterCommit"})
void requiresNewWithFailingListenerRestoresOuterTransaction(String callback) {
this.manager.addListener(failingListener(callback, true));
TransactionTemplate inner = nestedTransaction(TransactionDefinition.PROPAGATION_REQUIRES_NEW);

this.transaction.executeWithoutResult(outer -> {
Object resource = TransactionSynchronizationManager.getResource(this.dataSource);
this.jdbc.update("insert into entries values (1)");
inner.executeWithoutResult(status -> {
assertThat(TransactionSynchronizationManager.getResource(this.dataSource)).isNotSameAs(resource);
this.jdbc.update("insert into entries values (2)");
});
assertThat(TransactionSynchronizationManager.getResource(this.dataSource)).isSameAs(resource);
assertThat(TransactionSynchronizationManager.isActualTransactionActive()).isTrue();
this.jdbc.update("insert into entries values (3)");
});

assertThat(this.jdbc.queryForList("select id from entries order by id", Integer.class)).containsExactly(1, 2, 3);
}

private TransactionTemplate nestedTransaction(int propagation) {
TransactionTemplate nested = new TransactionTemplate(this.manager);
nested.setName("inner");
nested.setPropagationBehavior(propagation);
return nested;
}

private static TransactionExecutionListener failingListener(String callback, boolean innerOnly) {
return mock(TransactionExecutionListener.class, invocation -> {
if (invocation.getMethod().getName().equals(callback)) {
TransactionExecution transaction = invocation.getArgument(0);
if (!innerOnly || "inner".equals(transaction.getTransactionName())) {
throw new IllegalStateException("Listener failure");
}
}
return null;
});
}

}
5 changes: 5 additions & 0 deletions spring-tx/src/main/java/org/springframework/transaction/TransactionExecutionListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,11 @@
* {@link TransactionStatus} (for a {@link PlatformTransactionManager} transaction) or
* a {@link ReactiveTransaction} (for a {@link ReactiveTransactionManager} transaction).
*
* <p>Any exception or error thrown by a listener is logged and not propagated by
* {@link org.springframework.transaction.support.AbstractPlatformTransactionManager}
* or {@link org.springframework.transaction.reactive.AbstractReactiveTransactionManager}.
* Transaction processing and notifications to subsequent listeners continue.
*
* @author Juergen Hoeller
* @since 6.1
* @see ConfigurableTransactionManager#addListener
Expand Down
43 changes: 28 additions & 15 deletions ...ain/java/org/springframework/transaction/reactive/AbstractReactiveTransactionManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import java.util.List;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Predicate;

import org.apache.commons.logging.Log;
Expand Down Expand Up @@ -157,12 +158,12 @@ else if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUI
GenericReactiveTransaction status = newReactiveTransaction(
nestedSynchronizationManager, def, transaction, true,
false, debugEnabled, suspendedResources.orElse(null));
this.transactionExecutionListeners.forEach(listener -> listener.beforeBegin(status));
invokeExecutionListeners("beforeBegin", listener -> listener.beforeBegin(status));
return doBegin(nestedSynchronizationManager, transaction, def)
.doOnSuccess(ignore -> prepareSynchronization(nestedSynchronizationManager, status, def))
.doOnError(ex -> this.transactionExecutionListeners.forEach(listener -> listener.afterBegin(status, ex)))
.doOnError(ex -> invokeExecutionListeners("afterBegin", listener -> listener.afterBegin(status, ex)))
.thenReturn(status);
}).doOnSuccess(status -> this.transactionExecutionListeners.forEach(listener -> listener.afterBegin(status, null)))
}).doOnSuccess(status -> invokeExecutionListeners("afterBegin", listener -> listener.afterBegin(status, null)))
.onErrorResume(ErrorPredicates.RUNTIME_OR_ERROR,
ex -> resume(nestedSynchronizationManager, null, suspendedResources.orElse(null))
.then(Mono.error(ex)));
Expand Down Expand Up @@ -211,12 +212,12 @@ private Mono<ReactiveTransaction> handleExistingTransaction(TransactionSynchroni
return suspendedResources.flatMap(suspendedResourcesHolder -> {
GenericReactiveTransaction status = newReactiveTransaction(synchronizationManager,
definition, transaction, true, false, debugEnabled, suspendedResourcesHolder);
this.transactionExecutionListeners.forEach(listener -> listener.beforeBegin(status));
invokeExecutionListeners("beforeBegin", listener -> listener.beforeBegin(status));
return doBegin(synchronizationManager, transaction, definition)
.doOnSuccess(ignore -> prepareSynchronization(synchronizationManager, status, definition))
.doOnError(ex -> this.transactionExecutionListeners.forEach(listener -> listener.afterBegin(status, ex)))
.doOnError(ex -> invokeExecutionListeners("afterBegin", listener -> listener.afterBegin(status, ex)))
.thenReturn(status)
.doOnSuccess(ignore -> this.transactionExecutionListeners.forEach(listener -> listener.afterBegin(status, null)))
.doOnSuccess(ignore -> invokeExecutionListeners("afterBegin", listener -> listener.afterBegin(status, null)))
.onErrorResume(ErrorPredicates.RUNTIME_OR_ERROR, beginEx ->
resumeAfterBeginException(synchronizationManager, transaction, suspendedResourcesHolder, beginEx)
.then(Mono.error(beginEx)));
Expand Down Expand Up @@ -465,7 +466,7 @@ private Mono<Void> processCommit(TransactionSynchronizationManager synchronizati
if (status.isDebug()) {
logger.debug("Initiating transaction commit");
}
this.transactionExecutionListeners.forEach(listener -> listener.beforeCommit(status));
invokeExecutionListeners("beforeCommit", listener -> listener.beforeCommit(status));
return doCommit(synchronizationManager, status);
}
return Mono.empty();
Expand All @@ -479,7 +480,7 @@ private Mono<Void> processCommit(TransactionSynchronizationManager synchronizati
result = triggerAfterCompletion(synchronizationManager, status, TransactionSynchronization.STATUS_ROLLED_BACK)
.then(Mono.defer(() -> {
if (status.isNewTransaction()) {
this.transactionExecutionListeners.forEach(listener -> listener.afterRollback(status, null));
invokeExecutionListeners("afterRollback", listener -> listener.afterRollback(status, null));
}
return propagateException;
}));
Expand All @@ -488,7 +489,7 @@ else if (ErrorPredicates.TRANSACTION_EXCEPTION.test(ex)) {
result = triggerAfterCompletion(synchronizationManager, status, TransactionSynchronization.STATUS_UNKNOWN)
.then(Mono.defer(() -> {
if (status.isNewTransaction()) {
this.transactionExecutionListeners.forEach(listener -> listener.afterCommit(status, ex));
invokeExecutionListeners("afterCommit", listener -> listener.afterCommit(status, ex));
}
return propagateException;
}));
Expand All @@ -508,7 +509,7 @@ else if (ErrorPredicates.RUNTIME_OR_ERROR.test(ex)) {
.then(triggerAfterCompletion(synchronizationManager, status, TransactionSynchronization.STATUS_COMMITTED))
.then(Mono.defer(() -> {
if (status.isNewTransaction()) {
this.transactionExecutionListeners.forEach(listener -> listener.afterCommit(status, null));
invokeExecutionListeners("afterCommit", listener -> listener.afterCommit(status, null));
}
return Mono.empty();
}))));
Expand Down Expand Up @@ -550,7 +551,7 @@ private Mono<Void> processRollback(TransactionSynchronizationManager synchroniza
if (status.isDebug()) {
logger.debug("Initiating transaction rollback");
}
this.transactionExecutionListeners.forEach(listener -> listener.beforeRollback(status));
invokeExecutionListeners("beforeRollback", listener -> listener.beforeRollback(status));
return doRollback(synchronizationManager, status);
}
else {
Expand All @@ -571,15 +572,15 @@ private Mono<Void> processRollback(TransactionSynchronizationManager synchroniza
triggerAfterCompletion(synchronizationManager, status, TransactionSynchronization.STATUS_UNKNOWN)
.then(Mono.defer(() -> {
if (status.isNewTransaction()) {
this.transactionExecutionListeners.forEach(listener -> listener.afterRollback(status, ex));
invokeExecutionListeners("afterRollback", listener -> listener.afterRollback(status, ex));
}
return Mono.empty();
}))
.then(Mono.error(ex)))
.then(Mono.defer(() -> triggerAfterCompletion(synchronizationManager, status, TransactionSynchronization.STATUS_ROLLED_BACK)))
.then(Mono.defer(() -> {
if (status.isNewTransaction()) {
this.transactionExecutionListeners.forEach(listener -> listener.afterRollback(status, null));
invokeExecutionListeners("afterRollback", listener -> listener.afterRollback(status, null));
}
return Mono.empty();
}))
Expand Down Expand Up @@ -615,18 +616,30 @@ else if (status.hasTransaction()) {
logger.error("Commit exception overridden by rollback exception", ex);
return triggerAfterCompletion(synchronizationManager, status, TransactionSynchronization.STATUS_UNKNOWN)
.then(Mono.defer(() -> {
this.transactionExecutionListeners.forEach(listener -> listener.afterRollback(status, rbex));
invokeExecutionListeners("afterRollback", listener -> listener.afterRollback(status, rbex));
return Mono.empty();
}))
.then(Mono.error(rbex));
}).then(Mono.defer(() -> triggerAfterCompletion(synchronizationManager, status, TransactionSynchronization.STATUS_ROLLED_BACK)))
.then(Mono.defer(() -> {
this.transactionExecutionListeners.forEach(listener -> listener.afterRollback(status, null));
invokeExecutionListeners("afterRollback", listener -> listener.afterRollback(status, null));
return Mono.empty();
}));
}


private void invokeExecutionListeners(String callback, Consumer<TransactionExecutionListener> action) {
for (TransactionExecutionListener listener : this.transactionExecutionListeners) {
try {
action.accept(listener);
}
catch (Throwable ex) {
logger.error("TransactionExecutionListener." + callback + " threw exception", ex);
}
}
}


/**
* Trigger {@code beforeCommit} callbacks.
* @param synchronizationManager the synchronization manager bound to the current transaction
Expand Down
Loading