Spring Data R2DBC Transaction

Overview:

In this tutorial, I would like to show you how we could achieve DB transaction with Spring Data R2DBC Transaction operator & also with @Transactional.

Transaction:

A database transaction is a unit of work, which could be made of up of 1 or more SQL statements, which is either completed or rolled back. For example, If we consider a bank application, one customer wants to transfer money to another customer, we need to deduct money from 1 account (say A) and then we also need to ensure that we add the deducted amount to the another account (say B).

We had deducted money from account (A) and some un-handled error occurs when we add the money to the account (B) , we might leave the DB in an inconsistent state and also money we had deducted is simply vanished!! So either both operation should complete successfully or we should revert any change we had done as part of transaction.

Aim of this tutorial is to show we could do using Spring Data R2DBC Transactional operator and by using @Transactional.

Sample Application:

Lets consider a simple bank application in which the users can deposit money. As and when the users deposit money, we will update the account balance and we will also log an entry in the deposit event table. There is a business rule which confirms that minimum deposit should be $100.

Lets try to implement this.

Project Setup:

Lets set up a Spring project with the following dependencies.

• Table
  • I create 2 tables as shown here and we insert some accounts.
  • I add the check constraint to ensure that we do not log the entries which are below 100.
  • I keep this file under src/main/resources with the name init.sql.

create table account (
    id INT auto_increment,
    user_name VARCHAR(50),
    balance INT
);

create table money_deposit_event (
    id INT auto_increment,
    account_number INT,
    amount INT,
    foreign key (account_number) references account(id),
    check (amount > 99) -- business rule says there should be min $100 deposit
);

insert into account (user_name, balance) values
     ('Tara', 0),
     ('Daisy', 0),
     ('Fredericka', 0),
     ('Brita', 0);

• Account Entity

@Data
@ToString
public class Account {

    @Id
    private Integer id;
    private String userName;
    private Integer balance;

}

• MoneyDepositEvent Entity

@Data
@ToString
@NoArgsConstructor
@AllArgsConstructor(staticName = "create")
public class MoneyDepositEvent {

    @Id
    private Integer id;
    private Integer accountNumber;
    private Integer amount;

}

• Repository

@Repository
public interface AccountRepository extends ReactiveCrudRepository<Account, Integer> {
}

@Repository
public interface MoneyDepositRepository extends ReactiveCrudRepository<MoneyDepositEvent, Integer> {
}

• Money Deposit Request – DTO

@Data
@ToString
@NoArgsConstructor
@AllArgsConstructor(staticName = "create")
public class DepositRequest {

    private int account;
    private int amount;

}

•  Service
  • In this service class , when we get the deposit request, we immediately update the user balance and then we add an entry into the event table.
  • You might ask, why we are NOT checking the request amount first!? I do this intentionally, just to simulate some SQL exception.

@Service
public class BankService {

    @Autowired
    private AccountRepository accountRepository;

    @Autowired
    private MoneyDepositRepository eventRepository;
    
    public Mono<Void> deposit(DepositRequest request){
        return this.accountRepository.findById(request.getAccount())
                    .doOnNext(ac -> ac.setBalance(ac.getBalance() + request.getAmount()))
                    .flatMap(this.accountRepository::save)
                    .thenReturn(toEvent(request))
                    .flatMap(this.eventRepository::save)
                    .doOnError(System.out::println)
                    .then();
    }

    // create money deposit event from request
    private MoneyDepositEvent toEvent(DepositRequest request){
        return MoneyDepositEvent.create(
                null,
                    request.getAccount(),
                    request.getAmount()
        );
    }

}

So far the set up looks good. Now lets try to access our service and deposit money. For that I am going to use Junit tests as shown below.

• We run the init.sql as part of BeforeAll.

@SpringBootTest
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class R2dbcTransactionApplicationTests {

    @Value("classpath:init.sql")
    private Resource initSql;

    @Autowired
    private R2dbcEntityTemplate entityTemplate;


    @BeforeAll
    public void initDB() throws IOException {
        String query = StreamUtils.copyToString(initSql.getInputStream(), StandardCharsets.UTF_8);
        Mono<Void> mono = this.entityTemplate
                .getDatabaseClient()
                .sql(query)
                .then();

        StepVerifier.create(mono)
                .verifyComplete();
    }
}

• Now lets try to add 2 tests.
  • Test 1 : Here we try to deposit $500 into account number 1.  It should be successful.
  • Test 2 : Here we deposit $99. Ideally the user account balance should remain the same ($500) as we have an explicit constraint not to deposit anything below $100. So the transaction should get reverted.

@Test
@Order(1)
void transactionSuccess() {
    DepositRequest request = DepositRequest.create(1, 500);
    Mono<Account> mono = this.bankService.deposit(request)
            .then(getAccountDetails(request));
    StepVerifier.create(mono)
            .expectNextMatches(ac -> ac.getBalance() == 500)
            .verifyComplete();
}

@Test
@Order(2)
void transactionFailure() {
    DepositRequest request = DepositRequest.create(1, 99);
    Mono<Account> mono = this.bankService.deposit(request)
            .onErrorResume(ex -> Mono.empty())
            .then(getAccountDetails(request));
    StepVerifier.create(mono)
            .expectNextMatches(ac -> ac.getBalance() == 500) // no change
            .verifyComplete();
}

private Mono<Account> getAccountDetails(DepositRequest request){
    return this.repository.findById(request.getAccount())
            .doOnNext(System.out::println);
}

But if I run the test, I could see that my implementation has issues. It does through the exception because of the constraint which is expected. However the account balance has increased to 599 from 500 which should NOT be.

Account(id=1, userName=Tara, balance=500)
org.springframework.dao.DataIntegrityViolationException: executeMany; SQL [INSERT INTO money_deposit_event (account_number, amount) VALUES ($1, $2)]; Check constraint violation: "CONSTRAINT_FE: (AMOUNT > 99)"; SQL statement:
INSERT INTO money_deposit_event (account_number, amount) VALUES ($1, $2) [23513-200]; nested exception is io.r2dbc.spi.R2dbcDataIntegrityViolationException: [23513] [23513] Check constraint violation: "CONSTRAINT_FE: (AMOUNT > 99)"; SQL statement:
INSERT INTO money_deposit_event (account_number, amount) VALUES ($1, $2) [23513-200]
Account(id=1, userName=Tara, balance=599)

Spring Data R2DBC Transaction – @Transactional:

As you would have guessed, the problem can be easily solved by adding @Transactional on the method which does the deposit.

• The annotation ensures that the changes are NOT committed when there is an error signal in the pipeline.

@Transactional
public Mono<Void> deposit(DepositRequest request){
    return this.accountRepository.findById(request.getAccount())
                .doOnNext(ac -> ac.setBalance(ac.getBalance() + request.getAmount()))
                .flatMap(this.accountRepository::save)
                .thenReturn(toEvent(request))
                .flatMap(this.eventRepository::save)
                .doOnError(System.out::println)
                .then();
}

Rerun the same test and check the output. The tests will pass now.

Account(id=1, userName=Tara, balance=500)
org.springframework.dao.DataIntegrityViolationException: executeMany; SQL [INSERT INTO money_deposit_event (account_number, amount) VALUES ($1, $2)]; Check constraint violation: "CONSTRAINT_FE: (AMOUNT > 99)"; SQL statement:
INSERT INTO money_deposit_event (account_number, amount) VALUES ($1, $2) [23513-200]; nested exception is io.r2dbc.spi.R2dbcDataIntegrityViolationException: [23513] [23513] Check constraint violation: "CONSTRAINT_FE: (AMOUNT > 99)"; SQL statement:
INSERT INTO money_deposit_event (account_number, amount) VALUES ($1, $2) [23513-200]
Account(id=1, userName=Tara, balance=500)

Limitations of @Transactional:

There are some limitations of using this annotation.

• @Transactional method should be a public method
• It will work for external calls – only when any other class (object) calling this public method from this service class.
• It will NOT work for internal calls – if any other method within this service class calls this method.

If you do not like this approach, there is another way.

Spring Data R2DBC Transaction – Transactional Operator:

We can also use Transactional Operator as shown here.

@Autowired
private TransactionalOperator operator;

public Mono<Void> deposit(DepositRequest request){
    return this.accountRepository.findById(request.getAccount())
            .doOnNext(ac -> ac.setBalance(ac.getBalance() + request.getAmount()))
            .flatMap(this.accountRepository::save)
            .thenReturn(toEvent(request))
            .flatMap(this.eventRepository::save)
            .doOnError(System.out::println)
            .as(operator::transactional) // add this
            .then();
}

If we rerun the test, we can confirm that transactional operator ensures that it revert the changes automatically in case of errors.

Summary:

We were able to successfully demonstrate DB Transaction by using Spring Data R2DBC Transaction Operator and also with @Transactional annotation.

The source code is available here.

Learn more about Spring Data R2DBC.

• Spring Data R2DBC Query By Example
• Spring Data R2DBC Pagination
• Spring Data R2DBC Entity Callback

Happy coding 🙂