Introduce TransactionExecutionListener with begin/commit/rollback notifications

Includes ConfigurableTransactionManager interface for listener registration.
Includes additional introspection methods on TransactionExecution interface.
Includes default method declarations on TransactionStatus/SmartTransactionObject.

Closes gh-27479
This commit is contained in:
Juergen Hoeller
2023-08-06 14:03:44 +02:00
parent eaf54b54c3
commit 3253d2de89
22 changed files with 1372 additions and 481 deletions

View File

@@ -0,0 +1,54 @@
/*
* Copyright 2002-2023 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.transaction;
import java.util.Collection;
/**
* Common configuration interface for transaction manager implementations.
* Provides registration facilities for {@link TransactionExecutionListener}.
*
* @author Juergen Hoeller
* @since 6.1
* @see PlatformTransactionManager
* @see ReactiveTransactionManager
*/
public interface ConfigurableTransactionManager extends TransactionManager {
/**
* Set the transaction execution listeners for begin/commit/rollback callbacks
* from this transaction manager.
* @see #addListener
*/
void setTransactionExecutionListeners(Collection<TransactionExecutionListener> listeners);
/**
* Return the registered transaction execution listeners for this transaction manager.
* @see #setTransactionExecutionListeners
*/
Collection<TransactionExecutionListener> getTransactionExecutionListeners();
/**
* Conveniently register the given listener for begin/commit/rollback callbacks
* from this transaction manager.
* @see #getTransactionExecutionListeners()
*/
default void addListener(TransactionExecutionListener listener) {
getTransactionExecutionListeners().add(listener);
}
}

View File

@@ -42,6 +42,7 @@ import org.springframework.lang.Nullable;
* @see org.springframework.transaction.support.TransactionTemplate
* @see org.springframework.transaction.interceptor.TransactionInterceptor
* @see org.springframework.transaction.ReactiveTransactionManager
* @see ConfigurableTransactionManager
*/
public interface PlatformTransactionManager extends TransactionManager {
@@ -68,8 +69,7 @@ public interface PlatformTransactionManager extends TransactionManager {
* @see TransactionDefinition#getTimeout
* @see TransactionDefinition#isReadOnly
*/
TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
throws TransactionException;
TransactionStatus getTransaction(@Nullable TransactionDefinition definition) throws TransactionException;
/**
* Commit the given transaction, with regard to its status. If the transaction

View File

@@ -17,7 +17,7 @@
package org.springframework.transaction;
/**
* Representation of an ongoing reactive transaction.
* Representation of an ongoing {@link ReactiveTransactionManager} transaction.
* This is currently a marker interface extending {@link TransactionExecution}
* but may acquire further methods in a future revision.
*

View File

@@ -32,6 +32,7 @@ import org.springframework.lang.Nullable;
* @see org.springframework.transaction.reactive.TransactionalOperator
* @see org.springframework.transaction.interceptor.TransactionInterceptor
* @see org.springframework.transaction.PlatformTransactionManager
* @see ConfigurableTransactionManager
*/
public interface ReactiveTransactionManager extends TransactionManager {

View File

@@ -19,7 +19,8 @@ package org.springframework.transaction;
/**
* Common representation of the current state of a transaction.
* Serves as base interface for {@link TransactionStatus} as well as
* {@link ReactiveTransaction}.
* {@link ReactiveTransaction}, and as of 6.1 also as transaction
* representation for {@link TransactionExecutionListener}.
*
* @author Juergen Hoeller
* @since 5.2
@@ -27,29 +28,102 @@ package org.springframework.transaction;
public interface TransactionExecution {
/**
* Return whether the present transaction is new; otherwise participating
* in an existing transaction, or potentially not running in an actual
* transaction in the first place.
* Return the defined name of the transaction (possibly an empty String).
* <p>In case of Spring's declarative transactions, the exposed name will be
* the {@code fully-qualified class name + "." + method name} (by default).
* <p>The default implementation returns an empty String.
* @since 6.1
* @see TransactionDefinition#getName()
*/
boolean isNewTransaction();
default String getTransactionName() {
return "";
}
/**
* Return whether there is an actual transaction active: this is meant to cover
* a new transaction as well as participation in an existing transaction, only
* returning {@code false} when not running in an actual transaction at all.
* <p>The default implementation returns {@code true}.
* @since 6.1
* @see #isNewTransaction()
* @see #isNested()
* @see #isReadOnly()
*/
default boolean hasTransaction() {
return true;
}
/**
* Return whether the transaction manager considers the present transaction
* as new; otherwise participating in an existing transaction, or potentially
* not running in an actual transaction in the first place.
* <p>This is primarily here for transaction manager state handling.
* Prefer the use of {@link #hasTransaction()} for application purposes
* since this is usually semantically appropriate.
* <p>The "new" status can be transaction manager specific, e.g. returning
* {@code true} for an actual nested transaction but potentially {@code false}
* for a savepoint-based nested transaction scope if the savepoint management
* is explicitly exposed (such as on {@link TransactionStatus}). A combined
* check for any kind of nested execution is provided by {@link #isNested()}.
* <p>The default implementation returns {@code true}.
* @see #hasTransaction()
* @see #isNested()
* @see TransactionStatus#hasSavepoint()
*/
default boolean isNewTransaction() {
return true;
}
/**
* Return if this transaction executes in a nested fashion within another.
* <p>The default implementation returns {@code false}.
* @since 6.1
* @see #hasTransaction()
* @see #isNewTransaction()
* @see TransactionDefinition#PROPAGATION_NESTED
*/
default boolean isNested() {
return false;
}
/**
* Return if this transaction is defined as read-only transaction.
* <p>The default implementation returns {@code false}.
* @since 6.1
* @see TransactionDefinition#isReadOnly()
*/
default boolean isReadOnly() {
return false;
}
/**
* Set the transaction rollback-only. This instructs the transaction manager
* that the only possible outcome of the transaction may be a rollback, as
* alternative to throwing an exception which would in turn trigger a rollback.
* <p>The default implementation throws an UnsupportedOperationException.
* @see #isRollbackOnly()
*/
void setRollbackOnly();
default void setRollbackOnly() {
throw new UnsupportedOperationException("setRollbackOnly not supported");
}
/**
* Return whether the transaction has been marked as rollback-only
* (either by the application or by the transaction infrastructure).
* <p>The default implementation returns {@code false}.
* @see #setRollbackOnly()
*/
boolean isRollbackOnly();
default boolean isRollbackOnly() {
return false;
}
/**
* Return whether this transaction is completed, that is,
* whether it has already been committed or rolled back.
* <p>The default implementation returns {@code false}.
*/
boolean isCompleted();
default boolean isCompleted() {
return false;
}
}

View File

@@ -0,0 +1,88 @@
/*
* Copyright 2002-2023 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.transaction;
import org.springframework.lang.Nullable;
/**
* Callback interface for stateless listening to transaction creation/completion steps
* in a transaction manager. This is primarily meant for observation and statistics;
* consider stateful transaction synchronizations for resource management purposes.
*
* <p>In contrast to synchronizations, the transaction execution listener contract is
* commonly supported for thread-bound transactions as well as reactive transactions.
* The callback-provided {@link TransactionExecution} object will be either a
* {@link TransactionStatus} (for a {@link PlatformTransactionManager} transaction) or
* a {@link ReactiveTransaction} (for a {@link ReactiveTransactionManager} transaction).
*
* @author Juergen Hoeller
* @since 6.1
* @see ConfigurableTransactionManager#addListener
* @see org.springframework.transaction.support.TransactionSynchronizationManager#registerSynchronization
* @see org.springframework.transaction.reactive.TransactionSynchronizationManager#registerSynchronization
*/
public interface TransactionExecutionListener {
/**
* Callback before the transaction begin step.
* @param transaction the current transaction
*/
default void beforeBegin(TransactionExecution transaction) {
}
/**
* Callback after the transaction begin step.
* @param transaction the current transaction
* @param beginFailure an exception occurring during begin
* (or {@code null} after a successful begin step)
*/
default void afterBegin(TransactionExecution transaction, @Nullable Throwable beginFailure) {
}
/**
* Callback before the transaction commit step.
* @param transaction the current transaction
*/
default void beforeCommit(TransactionExecution transaction) {
}
/**
* Callback after the transaction commit step.
* @param transaction the current transaction
* @param commitFailure an exception occurring during commit
* (or {@code null} after a successful commit step)
*/
default void afterCommit(TransactionExecution transaction, @Nullable Throwable commitFailure) {
}
/**
* Callback before the transaction rollback step.
* @param transaction the current transaction
*/
default void beforeRollback(TransactionExecution transaction) {
}
/**
* Callback after the transaction rollback step.
* @param transaction the current transaction
* @param rollbackFailure an exception occurring during rollback
* (or {@code null} after a successful rollback step)
*/
default void afterRollback(TransactionExecution transaction, @Nullable Throwable rollbackFailure) {
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2023 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.
@@ -19,7 +19,8 @@ package org.springframework.transaction;
import java.io.Flushable;
/**
* Representation of the status of a transaction.
* Representation of an ongoing {@link PlatformTransactionManager} transaction.
* Extends the common {@link TransactionExecution} interface.
*
* <p>Transactional code can use this to retrieve status information,
* and to programmatically request a rollback (instead of throwing
@@ -44,12 +45,15 @@ public interface TransactionStatus extends TransactionExecution, SavepointManage
* <p>This method is mainly here for diagnostic purposes, alongside
* {@link #isNewTransaction()}. For programmatic handling of custom
* savepoints, use the operations provided by {@link SavepointManager}.
* <p>The default implementation returns {@code false}.
* @see #isNewTransaction()
* @see #createSavepoint()
* @see #rollbackToSavepoint(Object)
* @see #releaseSavepoint(Object)
*/
boolean hasSavepoint();
default boolean hasSavepoint() {
return false;
}
/**
* Flush the underlying session to the datastore, if applicable:
@@ -58,8 +62,10 @@ public interface TransactionStatus extends TransactionExecution, SavepointManage
* transaction manager does not have a flush concept. A flush signal may
* get applied to the primary resource or to transaction synchronizations,
* depending on the underlying resource.
* <p>The default implementation is empty, considering flush as a no-op.
*/
@Override
void flush();
default void flush() {
}
}

View File

@@ -19,6 +19,8 @@ package org.springframework.transaction.reactive;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -30,12 +32,14 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.lang.Nullable;
import org.springframework.transaction.ConfigurableTransactionManager;
import org.springframework.transaction.IllegalTransactionStateException;
import org.springframework.transaction.InvalidTimeoutException;
import org.springframework.transaction.ReactiveTransaction;
import org.springframework.transaction.ReactiveTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionException;
import org.springframework.transaction.TransactionExecutionListener;
import org.springframework.transaction.TransactionSuspensionNotSupportedException;
import org.springframework.transaction.UnexpectedRollbackException;
@@ -77,10 +81,24 @@ import org.springframework.transaction.UnexpectedRollbackException;
* @see TransactionSynchronizationManager
*/
@SuppressWarnings("serial")
public abstract class AbstractReactiveTransactionManager implements ReactiveTransactionManager, Serializable {
public abstract class AbstractReactiveTransactionManager
implements ReactiveTransactionManager, ConfigurableTransactionManager, Serializable {
protected transient Log logger = LogFactory.getLog(getClass());
private Collection<TransactionExecutionListener> transactionExecutionListeners = new ArrayList<>();
@Override
public final void setTransactionExecutionListeners(Collection<TransactionExecutionListener> listeners) {
this.transactionExecutionListeners = listeners;
}
@Override
public final Collection<TransactionExecutionListener> getTransactionExecutionListeners() {
return this.transactionExecutionListeners;
}
//---------------------------------------------------------------------
// Implementation of ReactiveTransactionManager
@@ -99,8 +117,7 @@ public abstract class AbstractReactiveTransactionManager implements ReactiveTran
// Use defaults if no transaction definition given.
TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
return TransactionSynchronizationManager.forCurrentTransaction()
.flatMap(synchronizationManager -> {
return TransactionSynchronizationManager.forCurrentTransaction().flatMap(synchronizationManager -> {
Object transaction = doGetTransaction(synchronizationManager);
@@ -139,13 +156,16 @@ public abstract class AbstractReactiveTransactionManager implements ReactiveTran
return Mono.defer(() -> {
GenericReactiveTransaction status = newReactiveTransaction(
nestedSynchronizationManager, def, transaction, true,
debugEnabled, suspendedResources.orElse(null));
false, debugEnabled, suspendedResources.orElse(null));
this.transactionExecutionListeners.forEach(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)))
.thenReturn(status);
}).onErrorResume(ErrorPredicates.RUNTIME_OR_ERROR,
}).doOnSuccess(status -> this.transactionExecutionListeners.forEach(listener -> listener.afterBegin(status, null)))
.onErrorResume(ErrorPredicates.RUNTIME_OR_ERROR,
ex -> resume(nestedSynchronizationManager, null, suspendedResources.orElse(null))
.then(Mono.error(ex)));
.then(Mono.error(ex)));
}));
}
else {
@@ -190,9 +210,13 @@ public abstract class AbstractReactiveTransactionManager implements ReactiveTran
Mono<SuspendedResourcesHolder> suspendedResources = suspend(synchronizationManager, transaction);
return suspendedResources.flatMap(suspendedResourcesHolder -> {
GenericReactiveTransaction status = newReactiveTransaction(synchronizationManager,
definition, transaction, true, debugEnabled, suspendedResourcesHolder);
return doBegin(synchronizationManager, transaction, definition).doOnSuccess(ignore ->
prepareSynchronization(synchronizationManager, status, definition)).thenReturn(status)
definition, transaction, true, false, debugEnabled, suspendedResourcesHolder);
this.transactionExecutionListeners.forEach(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)))
.thenReturn(status)
.doOnSuccess(ignore -> this.transactionExecutionListeners.forEach(listener -> listener.afterBegin(status, null)))
.onErrorResume(ErrorPredicates.RUNTIME_OR_ERROR, beginEx ->
resumeAfterBeginException(synchronizationManager, transaction, suspendedResourcesHolder, beginEx)
.then(Mono.error(beginEx)));
@@ -205,7 +229,7 @@ public abstract class AbstractReactiveTransactionManager implements ReactiveTran
}
// Nested transaction through nested begin and commit/rollback calls.
GenericReactiveTransaction status = newReactiveTransaction(synchronizationManager,
definition, transaction, true, debugEnabled, null);
definition, transaction, true, true, debugEnabled, null);
return doBegin(synchronizationManager, transaction, definition).doOnSuccess(ignore ->
prepareSynchronization(synchronizationManager, status, definition)).thenReturn(status);
}
@@ -228,7 +252,7 @@ public abstract class AbstractReactiveTransactionManager implements ReactiveTran
@Nullable Object transaction, boolean newTransaction, boolean debug, @Nullable Object suspendedResources) {
GenericReactiveTransaction status = newReactiveTransaction(synchronizationManager,
definition, transaction, newTransaction, debug, suspendedResources);
definition, transaction, newTransaction, false, debug, suspendedResources);
prepareSynchronization(synchronizationManager, status, definition);
return status;
}
@@ -238,11 +262,12 @@ public abstract class AbstractReactiveTransactionManager implements ReactiveTran
*/
private GenericReactiveTransaction newReactiveTransaction(
TransactionSynchronizationManager synchronizationManager, TransactionDefinition definition,
@Nullable Object transaction, boolean newTransaction, boolean debug, @Nullable Object suspendedResources) {
@Nullable Object transaction, boolean newTransaction, boolean nested, boolean debug,
@Nullable Object suspendedResources) {
return new GenericReactiveTransaction(transaction, newTransaction,
!synchronizationManager.isSynchronizationActive(),
definition.isReadOnly(), debug, suspendedResources);
return new GenericReactiveTransaction(definition.getName(), transaction,
newTransaction, !synchronizationManager.isSynchronizationActive(),
nested, definition.isReadOnly(), debug, suspendedResources);
}
/**
@@ -438,6 +463,7 @@ public abstract class AbstractReactiveTransactionManager implements ReactiveTran
if (status.isDebug()) {
logger.debug("Initiating transaction commit");
}
this.transactionExecutionListeners.forEach(listener -> listener.beforeCommit(status));
return doCommit(synchronizationManager, status);
}
return Mono.empty();
@@ -449,11 +475,21 @@ public abstract class AbstractReactiveTransactionManager implements ReactiveTran
Mono<Void> result = propagateException;
if (ErrorPredicates.UNEXPECTED_ROLLBACK.test(ex)) {
result = triggerAfterCompletion(synchronizationManager, status, TransactionSynchronization.STATUS_ROLLED_BACK)
.then(propagateException);
.then(Mono.defer(() -> {
if (status.isNewTransaction()) {
this.transactionExecutionListeners.forEach(listener -> listener.afterRollback(status, null));
}
return propagateException;
}));
}
else if (ErrorPredicates.TRANSACTION_EXCEPTION.test(ex)) {
result = triggerAfterCompletion(synchronizationManager, status, TransactionSynchronization.STATUS_UNKNOWN)
.then(propagateException);
.then(Mono.defer(() -> {
if (status.isNewTransaction()) {
this.transactionExecutionListeners.forEach(listener -> listener.afterCommit(status, ex));
}
return propagateException;
}));
}
else if (ErrorPredicates.RUNTIME_OR_ERROR.test(ex)) {
Mono<Void> mono;
@@ -471,7 +507,13 @@ public abstract class AbstractReactiveTransactionManager implements ReactiveTran
})
.then(Mono.defer(() -> triggerAfterCommit(synchronizationManager, status).onErrorResume(ex ->
triggerAfterCompletion(synchronizationManager, status, TransactionSynchronization.STATUS_COMMITTED).then(Mono.error(ex)))
.then(triggerAfterCompletion(synchronizationManager, status, TransactionSynchronization.STATUS_COMMITTED))));
.then(triggerAfterCompletion(synchronizationManager, status, TransactionSynchronization.STATUS_COMMITTED))
.then(Mono.defer(() -> {
if (status.isNewTransaction()) {
this.transactionExecutionListeners.forEach(listener -> listener.afterCommit(status, null));
}
return Mono.empty();
}))));
return commit
.onErrorResume(ex -> cleanupAfterCompletion(synchronizationManager, status)
@@ -510,6 +552,7 @@ public abstract class AbstractReactiveTransactionManager implements ReactiveTran
if (status.isDebug()) {
logger.debug("Initiating transaction rollback");
}
this.transactionExecutionListeners.forEach(listener -> listener.beforeRollback(status));
return doRollback(synchronizationManager, status);
}
else {
@@ -528,10 +571,22 @@ public abstract class AbstractReactiveTransactionManager implements ReactiveTran
}
})).onErrorResume(ErrorPredicates.RUNTIME_OR_ERROR, ex -> triggerAfterCompletion(
synchronizationManager, status, TransactionSynchronization.STATUS_UNKNOWN)
.then(Mono.error(ex)))
.then(Mono.defer(() -> {
if (status.isNewTransaction()) {
this.transactionExecutionListeners.forEach(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));
}
return Mono.empty();
}))
.onErrorResume(ex -> cleanupAfterCompletion(synchronizationManager, status).then(Mono.error(ex)))
.then(cleanupAfterCompletion(synchronizationManager, status));
.then(cleanupAfterCompletion(synchronizationManager, status));
}
/**
@@ -561,8 +616,16 @@ public abstract class AbstractReactiveTransactionManager implements ReactiveTran
}).onErrorResume(ErrorPredicates.RUNTIME_OR_ERROR, rbex -> {
logger.error("Commit exception overridden by rollback exception", ex);
return triggerAfterCompletion(synchronizationManager, status, TransactionSynchronization.STATUS_UNKNOWN)
.then(Mono.error(rbex));
}).then(triggerAfterCompletion(synchronizationManager, status, TransactionSynchronization.STATUS_ROLLED_BACK));
.then(Mono.defer(() -> {
this.transactionExecutionListeners.forEach(listener -> listener.afterRollback(status, rbex));
return Mono.empty();
}))
.then(Mono.error(rbex));
}).then(triggerAfterCompletion(synchronizationManager, status, TransactionSynchronization.STATUS_ROLLED_BACK))
.then(Mono.defer(() -> {
this.transactionExecutionListeners.forEach(listener -> listener.afterRollback(status, null));
return Mono.empty();
}));
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2023 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.
@@ -40,6 +40,9 @@ import org.springframework.util.Assert;
*/
public class GenericReactiveTransaction implements ReactiveTransaction {
@Nullable
private final String transactionName;
@Nullable
private final Object transaction;
@@ -47,6 +50,8 @@ public class GenericReactiveTransaction implements ReactiveTransaction {
private final boolean newSynchronization;
private final boolean nested;
private final boolean readOnly;
private final boolean debug;
@@ -61,6 +66,7 @@ public class GenericReactiveTransaction implements ReactiveTransaction {
/**
* Create a new {@code DefaultReactiveTransactionStatus} instance.
* @param transactionName the defined name of the transaction
* @param transaction underlying transaction object that can hold state
* for the internal transaction implementation
* @param newTransaction if the transaction is new, otherwise participating
@@ -73,19 +79,35 @@ public class GenericReactiveTransaction implements ReactiveTransaction {
* debug logging should be enabled.
* @param suspendedResources a holder for resources that have been suspended
* for this transaction, if any
* @since 6.1
*/
public GenericReactiveTransaction(
@Nullable Object transaction, boolean newTransaction, boolean newSynchronization,
boolean readOnly, boolean debug, @Nullable Object suspendedResources) {
@Nullable String transactionName, @Nullable Object transaction, boolean newTransaction,
boolean newSynchronization, boolean nested, boolean readOnly, boolean debug,
@Nullable Object suspendedResources) {
this.transactionName = transactionName;
this.transaction = transaction;
this.newTransaction = newTransaction;
this.newSynchronization = newSynchronization;
this.nested = nested;
this.readOnly = readOnly;
this.debug = debug;
this.suspendedResources = suspendedResources;
}
@Deprecated(since = "6.1", forRemoval = true)
public GenericReactiveTransaction(@Nullable Object transaction, boolean newTransaction,
boolean newSynchronization, boolean readOnly, boolean debug, @Nullable Object suspendedResources) {
this(null, transaction, newTransaction, newSynchronization, false, readOnly, debug, suspendedResources);
}
@Override
public String getTransactionName() {
return (this.transactionName != null ? this.transactionName : "");
}
/**
* Return the underlying transaction object.
@@ -96,9 +118,7 @@ public class GenericReactiveTransaction implements ReactiveTransaction {
return this.transaction;
}
/**
* Return whether there is an actual transaction active.
*/
@Override
public boolean hasTransaction() {
return (this.transaction != null);
}
@@ -109,16 +129,18 @@ public class GenericReactiveTransaction implements ReactiveTransaction {
}
/**
* Return if a new transaction synchronization has been opened
* for this transaction.
* Return if a new transaction synchronization has been opened for this transaction.
*/
public boolean isNewSynchronization() {
return this.newSynchronization;
}
/**
* Return if this transaction is defined as read-only transaction.
*/
@Override
public boolean isNested() {
return this.nested;
}
@Override
public boolean isReadOnly() {
return this.readOnly;
}
@@ -143,6 +165,9 @@ public class GenericReactiveTransaction implements ReactiveTransaction {
@Override
public void setRollbackOnly() {
if (this.completed) {
throw new IllegalStateException("Transaction completed");
}
this.rollbackOnly = true;
}

View File

@@ -19,6 +19,8 @@ package org.springframework.transaction.support;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
@@ -26,12 +28,14 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.lang.Nullable;
import org.springframework.transaction.ConfigurableTransactionManager;
import org.springframework.transaction.IllegalTransactionStateException;
import org.springframework.transaction.InvalidTimeoutException;
import org.springframework.transaction.NestedTransactionNotSupportedException;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionException;
import org.springframework.transaction.TransactionExecutionListener;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.TransactionSuspensionNotSupportedException;
import org.springframework.transaction.UnexpectedRollbackException;
@@ -82,7 +86,8 @@ import org.springframework.util.Assert;
* @see org.springframework.transaction.jta.JtaTransactionManager
*/
@SuppressWarnings("serial")
public abstract class AbstractPlatformTransactionManager implements PlatformTransactionManager, Serializable {
public abstract class AbstractPlatformTransactionManager
implements PlatformTransactionManager, ConfigurableTransactionManager, Serializable {
/**
* Always activate transaction synchronization, even for "empty" transactions
@@ -136,6 +141,8 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
private boolean rollbackOnCommitFailure = false;
private Collection<TransactionExecutionListener> transactionExecutionListeners = new ArrayList<>();
/**
* Set the transaction synchronization by the name of the corresponding constant
@@ -339,6 +346,16 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
return this.rollbackOnCommitFailure;
}
@Override
public final void setTransactionExecutionListeners(Collection<TransactionExecutionListener> listeners) {
this.transactionExecutionListeners = listeners;
}
@Override
public final Collection<TransactionExecutionListener> getTransactionExecutionListeners() {
return this.transactionExecutionListeners;
}
//---------------------------------------------------------------------
// Implementation of PlatformTransactionManager
@@ -385,7 +402,7 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
logger.debug("Creating new transaction with name [" + def.getName() + "]: " + def);
}
try {
return startTransaction(def, transaction, debugEnabled, suspendedResources);
return startTransaction(def, transaction, false, debugEnabled, suspendedResources);
}
catch (RuntimeException | Error ex) {
resume(null, suspendedResources);
@@ -403,20 +420,6 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
}
}
/**
* Start a new transaction.
*/
private TransactionStatus startTransaction(TransactionDefinition definition, Object transaction,
boolean debugEnabled, @Nullable SuspendedResourcesHolder suspendedResources) {
boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
DefaultTransactionStatus status = newTransactionStatus(
definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
doBegin(transaction, definition);
prepareSynchronization(status, definition);
return status;
}
/**
* Create a TransactionStatus for an existing transaction.
*/
@@ -446,7 +449,7 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
}
SuspendedResourcesHolder suspendedResources = suspend(transaction);
try {
return startTransaction(definition, transaction, debugEnabled, suspendedResources);
return startTransaction(definition, transaction, false, debugEnabled, suspendedResources);
}
catch (RuntimeException | Error beginEx) {
resumeAfterBeginException(transaction, suspendedResources, beginEx);
@@ -467,16 +470,24 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
// Create savepoint within existing Spring-managed transaction,
// through the SavepointManager API implemented by TransactionStatus.
// Usually uses JDBC 3.0 savepoints. Never activates Spring synchronization.
DefaultTransactionStatus status =
prepareTransactionStatus(definition, transaction, false, false, debugEnabled, null);
status.createAndHoldSavepoint();
DefaultTransactionStatus status = newTransactionStatus(
definition, transaction, false, false, true, debugEnabled, null);
this.transactionExecutionListeners.forEach(listener -> listener.beforeBegin(status));
try {
status.createAndHoldSavepoint();
}
catch (RuntimeException | Error ex) {
this.transactionExecutionListeners.forEach(listener -> listener.afterBegin(status, ex));
throw ex;
}
this.transactionExecutionListeners.forEach(listener -> listener.afterBegin(status, null));
return status;
}
else {
// Nested transaction through nested begin and commit/rollback calls.
// Usually only for JTA: Spring synchronization might get activated here
// in case of a pre-existing JTA transaction.
return startTransaction(definition, transaction, debugEnabled, null);
return startTransaction(definition, transaction, true, debugEnabled, null);
}
}
@@ -506,18 +517,40 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
return prepareTransactionStatus(definition, transaction, false, newSynchronization, debugEnabled, null);
}
/**
* Start a new transaction.
*/
private TransactionStatus startTransaction(TransactionDefinition definition, Object transaction,
boolean nested, boolean debugEnabled, @Nullable SuspendedResourcesHolder suspendedResources) {
boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
DefaultTransactionStatus status = newTransactionStatus(
definition, transaction, true, newSynchronization, nested, debugEnabled, suspendedResources);
this.transactionExecutionListeners.forEach(listener -> listener.beforeBegin(status));
try {
doBegin(transaction, definition);
}
catch (RuntimeException | Error ex) {
this.transactionExecutionListeners.forEach(listener -> listener.afterBegin(status, ex));
throw ex;
}
prepareSynchronization(status, definition);
this.transactionExecutionListeners.forEach(listener -> listener.afterBegin(status, null));
return status;
}
/**
* Create a new TransactionStatus for the given arguments,
* also initializing transaction synchronization as appropriate.
* @see #newTransactionStatus
* @see #prepareTransactionStatus
*/
protected final DefaultTransactionStatus prepareTransactionStatus(
private DefaultTransactionStatus prepareTransactionStatus(
TransactionDefinition definition, @Nullable Object transaction, boolean newTransaction,
boolean newSynchronization, boolean debug, @Nullable Object suspendedResources) {
DefaultTransactionStatus status = newTransactionStatus(
definition, transaction, newTransaction, newSynchronization, debug, suspendedResources);
definition, transaction, newTransaction, newSynchronization, false, debug, suspendedResources);
prepareSynchronization(status, definition);
return status;
}
@@ -525,21 +558,20 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
/**
* Create a TransactionStatus instance for the given arguments.
*/
protected DefaultTransactionStatus newTransactionStatus(
private DefaultTransactionStatus newTransactionStatus(
TransactionDefinition definition, @Nullable Object transaction, boolean newTransaction,
boolean newSynchronization, boolean debug, @Nullable Object suspendedResources) {
boolean newSynchronization, boolean nested, boolean debug, @Nullable Object suspendedResources) {
boolean actualNewSynchronization = newSynchronization &&
!TransactionSynchronizationManager.isSynchronizationActive();
return new DefaultTransactionStatus(
transaction, newTransaction, actualNewSynchronization,
definition.isReadOnly(), debug, suspendedResources);
return new DefaultTransactionStatus(definition.getName(), transaction, newTransaction,
actualNewSynchronization, nested, definition.isReadOnly(), debug, suspendedResources);
}
/**
* Initialize transaction synchronization as appropriate.
*/
protected void prepareSynchronization(DefaultTransactionStatus status, TransactionDefinition definition) {
private void prepareSynchronization(DefaultTransactionStatus status, TransactionDefinition definition) {
if (status.isNewSynchronization()) {
TransactionSynchronizationManager.setActualTransactionActive(status.hasTransaction());
TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(
@@ -734,6 +766,7 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
private void processCommit(DefaultTransactionStatus status) throws TransactionException {
try {
boolean beforeCompletionInvoked = false;
boolean commitListenerInvoked = false;
try {
boolean unexpectedRollback = false;
@@ -747,6 +780,8 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
logger.debug("Releasing transaction savepoint");
}
unexpectedRollback = status.isGlobalRollbackOnly();
this.transactionExecutionListeners.forEach(listener -> listener.beforeCommit(status));
commitListenerInvoked = true;
status.releaseHeldSavepoint();
}
else if (status.isNewTransaction()) {
@@ -754,6 +789,8 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
logger.debug("Initiating transaction commit");
}
unexpectedRollback = status.isGlobalRollbackOnly();
this.transactionExecutionListeners.forEach(listener -> listener.beforeCommit(status));
commitListenerInvoked = true;
doCommit(status);
}
else if (isFailEarlyOnGlobalRollbackOnly()) {
@@ -768,17 +805,19 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
}
}
catch (UnexpectedRollbackException ex) {
// can only be caused by doCommit
triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
this.transactionExecutionListeners.forEach(listener -> listener.afterRollback(status, null));
throw ex;
}
catch (TransactionException ex) {
// can only be caused by doCommit
if (isRollbackOnCommitFailure()) {
doRollbackOnCommitException(status, ex);
}
else {
triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
if (commitListenerInvoked) {
this.transactionExecutionListeners.forEach(listener -> listener.afterCommit(status, ex));
}
}
throw ex;
}
@@ -797,6 +836,9 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
}
finally {
triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
if (commitListenerInvoked) {
this.transactionExecutionListeners.forEach(listener -> listener.afterCommit(status, null));
}
}
}
@@ -832,6 +874,7 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
private void processRollback(DefaultTransactionStatus status, boolean unexpected) {
try {
boolean unexpectedRollback = unexpected;
boolean rollbackListenerInvoked = false;
try {
triggerBeforeCompletion(status);
@@ -840,12 +883,16 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
if (status.isDebug()) {
logger.debug("Rolling back transaction to savepoint");
}
this.transactionExecutionListeners.forEach(listener -> listener.beforeRollback(status));
rollbackListenerInvoked = true;
status.rollbackToHeldSavepoint();
}
else if (status.isNewTransaction()) {
if (status.isDebug()) {
logger.debug("Initiating transaction rollback");
}
this.transactionExecutionListeners.forEach(listener -> listener.beforeRollback(status));
rollbackListenerInvoked = true;
doRollback(status);
}
else {
@@ -874,10 +921,16 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
}
catch (RuntimeException | Error ex) {
triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
if (rollbackListenerInvoked) {
this.transactionExecutionListeners.forEach(listener -> listener.afterRollback(status, ex));
}
throw ex;
}
triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
if (rollbackListenerInvoked) {
this.transactionExecutionListeners.forEach(listener -> listener.afterRollback(status, null));
}
// Raise UnexpectedRollbackException if we had a global rollback-only marker
if (unexpectedRollback) {
@@ -915,9 +968,11 @@ public abstract class AbstractPlatformTransactionManager implements PlatformTran
catch (RuntimeException | Error rbex) {
logger.error("Commit exception overridden by rollback exception", ex);
triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
this.transactionExecutionListeners.forEach(listener -> listener.afterRollback(status, rbex));
throw rbex;
}
triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
this.transactionExecutionListeners.forEach(listener -> listener.afterRollback(status, null));
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2023 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.
@@ -60,6 +60,9 @@ public abstract class AbstractTransactionStatus implements TransactionStatus {
@Override
public void setRollbackOnly() {
if (this.completed) {
throw new IllegalStateException("Transaction completed");
}
this.rollbackOnly = true;
}
@@ -216,16 +219,4 @@ public abstract class AbstractTransactionStatus implements TransactionStatus {
throw new NestedTransactionNotSupportedException("This transaction does not support savepoints");
}
//---------------------------------------------------------------------
// Flushing support
//---------------------------------------------------------------------
/**
* This implementation is empty, considering flush as a no-op.
*/
@Override
public void flush() {
}
}

View File

@@ -50,6 +50,9 @@ import org.springframework.util.Assert;
*/
public class DefaultTransactionStatus extends AbstractTransactionStatus {
@Nullable
private final String transactionName;
@Nullable
private final Object transaction;
@@ -57,6 +60,8 @@ public class DefaultTransactionStatus extends AbstractTransactionStatus {
private final boolean newSynchronization;
private final boolean nested;
private final boolean readOnly;
private final boolean debug;
@@ -67,6 +72,7 @@ public class DefaultTransactionStatus extends AbstractTransactionStatus {
/**
* Create a new {@code DefaultTransactionStatus} instance.
* @param transactionName the defined name of the transaction
* @param transaction underlying transaction object that can hold state
* for the internal transaction implementation
* @param newTransaction if the transaction is new, otherwise participating
@@ -79,19 +85,35 @@ public class DefaultTransactionStatus extends AbstractTransactionStatus {
* debug logging should be enabled.
* @param suspendedResources a holder for resources that have been suspended
* for this transaction, if any
* @since 6.1
*/
public DefaultTransactionStatus(
@Nullable Object transaction, boolean newTransaction, boolean newSynchronization,
boolean readOnly, boolean debug, @Nullable Object suspendedResources) {
@Nullable String transactionName, @Nullable Object transaction, boolean newTransaction,
boolean newSynchronization, boolean nested, boolean readOnly, boolean debug,
@Nullable Object suspendedResources) {
this.transactionName = transactionName;
this.transaction = transaction;
this.newTransaction = newTransaction;
this.newSynchronization = newSynchronization;
this.nested = nested;
this.readOnly = readOnly;
this.debug = debug;
this.suspendedResources = suspendedResources;
}
@Deprecated(since = "6.1", forRemoval = true)
public DefaultTransactionStatus(@Nullable Object transaction, boolean newTransaction,
boolean newSynchronization, boolean readOnly, boolean debug, @Nullable Object suspendedResources) {
this(null, transaction, newTransaction, newSynchronization, false, readOnly, debug, suspendedResources);
}
@Override
public String getTransactionName() {
return (this.transactionName != null ? this.transactionName : "");
}
/**
* Return the underlying transaction object.
@@ -102,9 +124,7 @@ public class DefaultTransactionStatus extends AbstractTransactionStatus {
return this.transaction;
}
/**
* Return whether there is an actual transaction active.
*/
@Override
public boolean hasTransaction() {
return (this.transaction != null);
}
@@ -115,16 +135,18 @@ public class DefaultTransactionStatus extends AbstractTransactionStatus {
}
/**
* Return if a new transaction synchronization has been opened
* for this transaction.
* Return if a new transaction synchronization has been opened for this transaction.
*/
public boolean isNewSynchronization() {
return this.newSynchronization;
}
/**
* Return if this transaction is defined as read-only transaction.
*/
@Override
public boolean isNested() {
return this.nested;
}
@Override
public boolean isReadOnly() {
return this.readOnly;
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2023 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.
@@ -23,29 +23,34 @@ import java.io.Flushable;
* return an internal rollback-only marker, typically from another
* transaction that has participated and marked it as rollback-only.
*
* <p>Autodetected by DefaultTransactionStatus, to always return a
* current rollbackOnly flag even if not resulting from the current
* <p>Autodetected by {@link DefaultTransactionStatus} in order to always
* return a current rollbackOnly flag even if not resulting from the current
* TransactionStatus.
*
* @author Juergen Hoeller
* @since 1.1
* @see DefaultTransactionStatus#isRollbackOnly
* @see DefaultTransactionStatus#isGlobalRollbackOnly()
*/
public interface SmartTransactionObject extends Flushable {
/**
* Return whether the transaction is internally marked as rollback-only.
* Can, for example, check the JTA UserTransaction.
* <p>The default implementation returns {@code false}.
* @see jakarta.transaction.UserTransaction#getStatus
* @see jakarta.transaction.Status#STATUS_MARKED_ROLLBACK
*/
boolean isRollbackOnly();
default boolean isRollbackOnly() {
return false;
}
/**
* Flush the underlying sessions to the datastore, if applicable:
* for example, all affected Hibernate/JPA sessions.
* <p>The default implementation is empty, considering flush as a no-op.
*/
@Override
void flush();
default void flush() {
}
}