how to limit the transaction scope with spring data reactive mongodb transactionalOperator

Question

I need to reduce the tx scope for performance reasons. for example

reactiveMongoTemplate.save(a)
    .then(
        reactiveMongoTemplate.inTransaction().execute(rmo -> methodInTx(rmo, trade))
    )
    .subscibe()

save(a) not participates the tx

ReactiveMongoTemplate.inTransaction() is deprecated since 2.2.

I had tried TransactionalOperator, it doesn't seem to support scoped transaction

    mongo.setSessionSynchronization(SessionSynchronization.ON_ACTUAL_TRANSACTION);
    txMongo.setSessionSynchronization(SessionSynchronization.ALWAYS);

    mongo.save(a, A.class)
            .flatMap(aa -> {
                aa.value = 1;
                return mongo.save(aa);
            })
            .flatMap(aa -> {
                aa.value = 2;
                return txMongo.save(aa).map(aaa -> {
                    if (true) throw new RuntimeException();
                    return aaa;
                });
            })
            .as(transactionalOperator::transactional)
            .doOnError(e -> e.printStackTrace()).subscribe();

I want value is set to 1, only 2 rolls back. But it seems transationalOperater started a transaction from the very beginning, mongo with session synchronization ON_ACTUAL_TRANSACTION also participates the transaction. How can I limit tx scope?

Answer 1

I turned out I should call transactionalOperator another way

    mongo.save(a, A.class)
        .flatMap(aa -> {
            aa.value = 1;
            return mongo.save(aa);
        })
        .flatMap(aa -> {
            aa.value = 2;
            return transactionalOperator.execute(s -> 
                mongo.save(aa).map(aaa -> {
                    if (true) throw new RuntimeException();
                    return aaa;
                })
           ).reduce((a, b) -> a);
        })
        .doOnError(e -> e.printStackTrace()).subscribe();