My project uses spring-data-mongodb
, everything is reactive. There is a bean with a transactional method using declarative transactions. The relevant code fragments follow:
@Configuration
public class Config {
@Bean
public ReactiveMongoTransactionManager reactiveMongoTransactionManager() {
return new ReactiveMongoTransactionManager(reactiveMongoDbFactory());
}
...
}
@Service
public class MyService {
private final ReactiveMongoOperations mongoOperations;
...
@Transactional
public Mono<User> saveUser(User user) {
return mongoOperations.insert(user).then(anotherInsertOnMongoOperations()).thenReturn(user);
}
}
Nothing unusual here.
I can see in logs that the transactions get started before documents get inserted, and after that they are committed:
DEBUG o.s.d.m.ReactiveMongoTransactionManager - About to start transaction for session [ClientSessionImpl@62de8058 id = {"id": {"$binary": "fye2h5JkRh6yL3MTqtC0Xw==", "$type": "04"}}, causallyConsistent = true, txActive = false, txNumber = 1, error = d != java.lang.Boolean].
DEBUG o.s.d.m.ReactiveMongoTransactionManager - Started transaction for session [ClientSessionImpl@62de8058 id = {"id": {"$binary": "fye2h5JkRh6yL3MTqtC0Xw==", "$type": "04"}}, causallyConsistent = true, txActive = true, txNumber = 2, error = d != java.lang.Boolean].
... inserts follow, then ...
DEBUG o.s.d.m.ReactiveMongoTransactionManager - Initiating transaction commit
DEBUG o.s.d.m.ReactiveMongoTransactionManager - About to commit transaction for session [ClientSessionImpl@62de8058 id = {"id": {"$binary": "fye2h5JkRh6yL3MTqtC0Xw==", "$type": "04"}}, causallyConsistent = true, txActive = true, txNumber = 2, error = d != java.lang.Boolean].
But sometimes, as I can see by the contents of the database, only the first of the inserts is persisted, and the second is lost. After trying to model the situation, I found that this 'loss' happens when the whole reactive pipeline gets cancelled (not every time, but I was able to produce a test that reproduced the situation with high probability).
I added .doOnSuccessOrError()
and .doOnCancel()
with some logging after the final operator of my method. In the 'normal' case (without a cancel), doOnSuccessOrError
logs successfully. But when a cancellation happens, sometimes the sequence of the events in the log is like this:
- A transaction is initiated
- An insert happens
- A cancel occurs
- nothing gets logged by the final
doOnSuccessOrError()
, and something is logged in theonCancel()
there (so the cancellation seems to happen 'right in the middle' of the execution of the business method) - ... but the transaction still gets committed!
TransactionAspectSupport.ReactiveTransactionSupport
contains the following code (used for this case):
return Mono.<Object, ReactiveTransactionInfo>usingWhen(
Mono.just(it),
txInfo -> {
try {
return (Mono<?>) invocation.proceedWithInvocation();
}
catch (Throwable ex) {
return Mono.error(ex);
}
},
this::commitTransactionAfterReturning,
(txInfo, err) -> Mono.empty(),
this::commitTransactionAfterReturning)
The last argument is onCancel
handler.
This means that on cancel, the transaction actually gets committed.
The question is: why? When a cancellation happens due to reasons external to a reactive pipeline, there is a probability that some operations inside a transaction have completed and some haven't (and never will). Committing at such a moment produces a partial commit which violates the atomicity requirement.
It seems more logical to initiate a rollback instead. But I suppose that the authors of spring-tx
did this choice on purpose. I wonder, what is the reason for this?
P.S. To verify my point, I patched spring-tx
5.2.3 (by the way, that's the version used by the project) so that the code looks like this:
return Mono.<Object, ReactiveTransactionInfo>usingWhen(
Mono.just(it),
txInfo -> {
try {
return (Mono<?>) invocation.proceedWithInvocation();
}
catch (Throwable ex) {
return Mono.error(ex);
}
},
this::commitTransactionAfterReturning,
(txInfo, err) -> Mono.empty(),
this::rollbackTransactionDueToCancel)
private Mono<Void> rollbackTransactionDueToCancel(@Nullable ReactiveTransactionInfo txInfo) {
if (txInfo != null && txInfo.getReactiveTransaction() != null) {
if (logger.isDebugEnabled()) {
logger.debug("Rolling transaction back for [" + txInfo.getJoinpointIdentification() + "] due to cancel");
}
return txInfo.getTransactionManager().rollback(txInfo.getReactiveTransaction());
}
return Mono.empty();
}
(basically, just changed the on-cancel behavior to rollback), and with this patch my tests do not produce any inconsistent data anymore.
saveUser
method?concatMap()
on aKafkaReceiver.receive()
fromreactor-kafka
. The cancellations occur due to retries (retryBackoff()
is used). But all this seemed to be irrelevant for the question.MyService
). That would lead to problems.retryBackoff()
behavior, so I don't have a stack trace (and it is not relevant here).