Table of Contents

Prioritize Oracle Transactions with Micronaut Data JDBC

Build an inventory example where an Oracle HIGH-priority checkout can roll back conflicting LOW-priority background work.

1. Getting Started

In this guide, we will create a Python application built with Pyronaut.

In this guide, you will build a small inventory application with Micronaut Data JDBC and Oracle Database. A slow stock reconciliation is background work, so it runs with LOW priority. A customer checkout needs the same last item and runs with HIGH priority. When Oracle priority transactions are enabled, Oracle rolls back the reconciliation once the checkout has waited long enough, so the customer does not wait for background work.

You will use @OracleTransactional to declare the priority of each transaction, and handle OracleTransactionPriorityException, which Micronaut Data throws when Oracle rolls back a lower-priority transaction.

2. What you will need

To complete this guide, you will need the following:

  • Some time on your hands

  • GraalPy installed and the Pyronaut CLI available locally

  • Docker installed to run Oracle Database with Micronaut Test Resources.

3. Solution

We recommend that you follow the instructions in the next sections and create the application step by step. However, you can go right to the completed example.

4. Writing the Application

Create an application using the Pyronaut CLI (Command Line Interface) or Pyronaut Launch

pyronaut create example.micronaut.micronautguide --features=data-jdbc,oracle,serialization-jackson,validation

The oracle feature adds the Oracle JDBC driver, com.oracle.database.jdbc:ojdbc11, to the runtime dependencies in pyproject.toml.

4.1. Database Configuration

Configure the datasource:

config/application.toml
[datasources.default]
schema-generate = "CREATE_DROP"
(1)
driver-class-name = "oracle.jdbc.OracleDriver"
(2)
db-type = "oracle"
(3)
dialect = "ORACLE"
1 Use Oracle driver.
2 In order for the database to be properly detected by Micronaut Test Resources.
3 Configure the Oracle dialect.

5. Configure Oracle Priority Transactions

Transaction priority is an Oracle Database feature. A database administrator decides how long a higher-priority transaction waits for a lower-priority one that holds a row lock, and whether Oracle then rolls back the blocker:

See Oracle’s Managing Transactions documentation for details.

For this guide, Micronaut Test Resources copies a startup script into the Oracle Database Free container:

config/application.toml
[test-resources.containers.oracle]
image-name = "gvenzl/oracle-free"
image-tag = "latest"
startup-timeout = "600s"
(1)
copy-to-container = [{ "config/oracle/priority-txns.sql" = "/container-entrypoint-startdb.d/priority-txns.sql" }]
1 Copy the script to the directory that the container runs on every database startup.
config/oracle/priority-txns.sql
WHENEVER SQLERROR EXIT SQL.SQLCODE

-- Set the CDB defaults first. This also covers a startup where the PDB has
-- not yet been opened or has no explicit parameter override.
ALTER SYSTEM SET priority_txns_high_wait_target=5 SCOPE=MEMORY;
ALTER SYSTEM SET priority_txns_medium_wait_target=10 SCOPE=MEMORY;
ALTER SYSTEM SET priority_txns_mode=ROLLBACK SCOPE=MEMORY;

-- The application connects to the default Oracle Free PDB. Set the parameters
-- in that PDB so the JDBC sessions used by the demo inherit them.
(1)
ALTER SESSION SET CONTAINER = FREEPDB1;

(2)
ALTER SYSTEM SET priority_txns_high_wait_target=5 SCOPE=MEMORY;
ALTER SYSTEM SET priority_txns_medium_wait_target=10 SCOPE=MEMORY;
(3)
ALTER SYSTEM SET priority_txns_mode=ROLLBACK SCOPE=MEMORY;

EXIT;
1 Set the defaults for the container database, then switch to FREEPDB1, the pluggable database that the application connects to.
2 The container runs the script as SYSDBA, so the application user does not need privileges to change instance parameters.
3 SCOPE=MEMORY keeps the settings for the lifetime of the throwaway container.

In other environments, a database administrator applies the same ALTER SYSTEM statements.

If you change the script while a Test Resources server is running, stop it with pyronaut test-resources-server stop so the next run starts a new container.

6. Inventory Item

Create an entity for the inventory item:

src/example/micronaut/inventory_item.py
from dataclasses import dataclass
from typing import Annotated

from micronaut.data.annotation import Id, MappedEntity
from micronaut.serde.annotation import Serdeable

from .status import Status


@Serdeable
@MappedEntity("inventory_item")
@dataclass
class InventoryItem:
    id: Annotated[int, Id]
    name: str
    available_quantity: int
    status: Status

Create the item status:

src/example/micronaut/status.py
from enum import Enum


class Status(Enum):
    AVAILABLE = "AVAILABLE"
    RECONCILED = "RECONCILED"
    CHECKED_OUT = "CHECKED_OUT"

7. Repository

Create a repository that locks the item while a transaction works with it:

src/example/micronaut/inventory_item_repository.py
from java.util import Optional
from micronaut.data.jdbc.annotation import JdbcRepository
from micronaut.data.model.query.builder.sql import Dialect
from micronaut.data.repository import CrudRepository

from .inventory_item import InventoryItem


@JdbcRepository(dialect=Dialect.ORACLE)  (1)
class InventoryItemRepository(CrudRepository[InventoryItem, int]):

    def findByIdForUpdate(self, id: int) -> Optional[InventoryItem]: ...  (2)
1 Micronaut Data generates SQL for the Oracle dialect at compilation time.
2 The ForUpdate suffix generates a SELECT …​ FOR UPDATE query. The row stays locked until the surrounding transaction ends.

8. Transactions with Different Priorities

Create a service with a LOW-priority reconciliation and a HIGH-priority checkout:

src/example/micronaut/inventory_service.py
import time

from jakarta.inject import Singleton
from micronaut.transaction.annotation import OracleTransactional, Transactional

from .inventory_item import InventoryItem
from .inventory_item_repository import InventoryItemRepository
from .status import Status

DEMO_ITEM_ID = 1


@Singleton
class InventoryService:

    def __init__(self, inventory_item_repository: InventoryItemRepository):
        self.inventory_item_repository = inventory_item_repository

    @Transactional
    def reset(self) -> InventoryItem:
        item = InventoryItem(DEMO_ITEM_ID, "Last available item", 1, Status.AVAILABLE)
        if self.inventory_item_repository.existsById(DEMO_ITEM_ID):
            return self.inventory_item_repository.update(item)
        return self.inventory_item_repository.save(item)

    @Transactional(readOnly=True)
    def find(self) -> InventoryItem:
        return self.inventory_item_repository.findById(DEMO_ITEM_ID).orElseThrow()

    @OracleTransactional(priority=OracleTransactional.Priority.LOW)  (1)
    def reconcile(self, count_seconds: int) -> InventoryItem:
        item = self.inventory_item_repository.findByIdForUpdate(DEMO_ITEM_ID).orElseThrow()  (2)
        time.sleep(count_seconds)  (3)
        return self.inventory_item_repository.update(
            InventoryItem(item.id, item.name, item.available_quantity, Status.RECONCILED))  (4)

    @OracleTransactional(priority=OracleTransactional.Priority.HIGH)  (5)
    def checkout(self) -> InventoryItem:
        item = self.inventory_item_repository.findByIdForUpdate(DEMO_ITEM_ID).orElseThrow()  (6)
        return self.inventory_item_repository.update(
            InventoryItem(item.id, item.name, 0, Status.CHECKED_OUT))
1 @OracleTransactional starts a transaction and sets the Oracle transaction priority for it. Micronaut Data restores the session priority when the transaction completes, so pooled connections do not keep it.
2 Lock the item. The lock alone makes the reconciliation a blocker for any other transaction that needs the row.
3 Simulate a slow count in an external system while the row is locked.
4 If Oracle rolled back the reconciliation while it was counting, this write is the first statement to find out. Micronaut Data rolls back the transaction and throws OracleTransactionPriorityException.
5 Checkout is the critical operation, so it runs with HIGH priority. HIGH is also the default for @OracleTransactional.
6 Checkout waits for the row lock. After the configured wait target, Oracle rolls back the reconciliation and checkout acquires the lock.

A priority rollback is a failed transaction; it is not retried automatically. The reconciliation read the item before checkout changed it, so replaying the same work could overwrite the checkout. To retry, run the operation in a new transaction that reads the item again and decides whether the work is still needed.

9. Controller

Expose the operations as HTTP endpoints:

src/example/micronaut/inventory_controller.py
from typing import Annotated

from jakarta.validation.constraints import Max, Min
from micronaut.http.annotation import Controller, Get, Post, QueryValue
from micronaut.scheduling import TaskExecutors
from micronaut.scheduling.annotation import ExecuteOn

from .inventory_item import InventoryItem
from .inventory_service import InventoryService


@ExecuteOn(TaskExecutors.BLOCKING)  (1)
@Controller("/inventory")
class InventoryController:

    def __init__(self, inventory_service: InventoryService):
        self.inventory_service = inventory_service

    @Get
    def item(self) -> InventoryItem:
        return self.inventory_service.find()

    @Post("/reset")
    def reset(self) -> InventoryItem:
        return self.inventory_service.reset()

    @Post("/reconcile")
    def reconcile(self, count_seconds: Annotated[
            int, QueryValue(value="countSeconds", defaultValue="20"), Min(1), Max(120)]) -> InventoryItem:  (2)
        return self.inventory_service.reconcile(count_seconds)

    @Post("/checkout")
    def checkout(self) -> InventoryItem:
        return self.inventory_service.checkout()
1 The operations block while they wait for row locks, so run them on the blocking executor instead of the event loop.
2 The reconciliation request stays open while its transaction holds the lock, so you can send a checkout from another request.

10. Handle a Priority Rollback

Without a handler, an OracleTransactionPriorityException results in a 500 Internal Server Error response. A priority rollback is an expected outcome of contention, so report it as a conflict instead:

src/example/micronaut/transaction_priority_exception_handler.py
from jakarta.inject import Singleton
from micronaut.http import HttpRequest, HttpResponse, HttpStatus
from micronaut.http.annotation import Produces
from micronaut.http.server.exceptions import ExceptionHandler
from micronaut.http.server.exceptions.response import ErrorContext, ErrorResponseProcessor
from micronaut.transaction.exceptions import OracleTransactionPriorityException


@Produces
@Singleton
class TransactionPriorityExceptionHandler(
        ExceptionHandler[OracleTransactionPriorityException, HttpResponse]):  (1)

    def __init__(self, error_response_processor: ErrorResponseProcessor):
        self.error_response_processor = error_response_processor

    def handle(self, request: HttpRequest, exception: OracleTransactionPriorityException) -> HttpResponse:
        error_context = (ErrorContext.builder(request)
                         .cause(exception)
                         .errorMessage("Oracle rolled back this operation in favor of a higher-priority transaction")  (2)
                         .build())
        return self.error_response_processor.processResponse(
            error_context, HttpResponse.status(HttpStatus.CONFLICT))  (3)
1 Handle OracleTransactionPriorityException. Micronaut Data throws it for both Oracle errors that report a priority rollback: ORA-63300 for the statement that was running when Oracle rolled back the transaction, and ORA-63302 for a later statement or commit.
2 Tell the client why the operation failed.
3 Respond with 409 Conflict, so the client can re-read the item and retry if the operation is still needed.

11. Test

Add a test that runs a reconciliation and a checkout against the same item:

tests/example/micronaut/test_inventory_controller.py
import time

import pytest
import requests
from java.sql import SQLException
from java.util.concurrent import CompletableFuture, TimeUnit
from micronaut.transaction import TransactionOperations
from pyronaut.test import MicronautTest, micronaut_test_fixture

from example.micronaut.inventory_service import DEMO_ITEM_ID

ORA_RESOURCE_BUSY = 54


@pytest.fixture
def my_context(request):
    fixture = micronaut_test_fixture(
        request,
        MicronautTest(transactional=False),  (1)
    )
    yield fixture
    fixture.stop()


@pytest.fixture
def client(my_context):
    client = requests.with_context(my_context)
    client.post("/inventory/reset")
    return client


def test_reconciliation_commits_without_contention(client):
    response = client.post("/inventory/reconcile?countSeconds=1")

    assert response.status_code == 200
    assert response.json()["status"] == "RECONCILED"
    assert response.json()["available_quantity"] == 1


def test_high_priority_checkout_rolls_back_low_priority_reconciliation(client, my_context):
    reconciliation = CompletableFuture.supplyAsync(
        lambda: client.post("/inventory/reconcile?countSeconds=8"))  (2)
    await_item_locked(my_context[TransactionOperations])  (3)

    checked_out = client.post("/inventory/checkout")  (4)
    assert checked_out.status_code == 200
    assert checked_out.json()["status"] == "CHECKED_OUT"

    rolled_back = reconciliation.get(15, TimeUnit.SECONDS)
    assert rolled_back.status_code == 409  (5)
    assert "Oracle rolled back this operation in favor of a higher-priority transaction" in rolled_back.text  (6)

    item = client.get("/inventory").json()
    assert item["status"] == "CHECKED_OUT"  (7)
    assert item["available_quantity"] == 0


def test_reconciliation_rejects_invalid_count_duration(client):
    response = client.post("/inventory/reconcile?countSeconds=0")

    assert response.status_code == 400  (8)


def await_item_locked(transaction_operations):
    deadline = time.monotonic() + 10
    while time.monotonic() < deadline:
        if is_item_locked(transaction_operations):
            return
        time.sleep(0.1)
    pytest.fail("The reconciliation did not lock the inventory item")


def is_item_locked(transaction_operations) -> bool:
    def probe(status) -> bool:
        statement = status.getConnection().prepareStatement(
            "SELECT id FROM inventory_item WHERE id = ? FOR UPDATE NOWAIT")
        try:
            statement.setLong(1, DEMO_ITEM_ID)
            statement.executeQuery().close()
            return False
        except SQLException as e:
            if e.getErrorCode() == ORA_RESOURCE_BUSY:
                return True
            raise
        finally:
            statement.close()

    return transaction_operations.executeWrite(probe)
1 Disable the test transaction so that each request runs its own transaction.
2 Send the reconciliation from another thread, because the request stays open while it holds the lock. The count takes eight seconds, longer than the five-second HIGH wait target in priority-txns.sql.
3 Wait until the reconciliation holds the row lock. The check uses NOWAIT, so it fails immediately with ORA-00054 instead of waiting for the lock.
4 Checkout waits for the lock, and Oracle rolls back the reconciliation after five seconds.
5 The reconciliation request fails with 409 Conflict from the exception handler.
6 The response body contains the message from the exception handler.
7 Only the checkout is committed.
8 Bean Validation rejects a countSeconds value outside the allowed range with 400 Bad Request, before any transaction starts.

12. Testing the Application

To run the tests:

pyronaut install
pyronaut validate-config
pyronaut test

When you run the tests, Micronaut Test Resources starts an Oracle Database Free container and runs priority-txns.sql before the application connects.

13. Run the Application

Start the application with pyronaut run, and reset the item:

curl -X POST http://localhost:8080/inventory/reset

In one terminal, start a reconciliation. The request stays open for 20 seconds while it holds the item lock:

curl -i -X POST http://localhost:8080/inventory/reconcile

Within those 20 seconds, check out the item from a second terminal:

curl -X POST http://localhost:8080/inventory/checkout

After about five seconds, the checkout responds with the item in the CHECKED_OUT status, and the reconciliation responds with 409 Conflict.

If you start the checkout without a running reconciliation, it completes immediately. If you run only the reconciliation, it commits after the count and responds with the RECONCILED status.

13.1. Micronaut Test Resources Goals

Micronaut Test Resources aims to achieve the following goals:

  • zero-configuration: without adding any configuration, test resources should be spawned and the application configured to use them. Configuration is only required for advanced use cases.

  • classpath isolation: use of test resources shouldn’t leak into your application classpath, nor your test classpath

  • compatible with GraalVM native: if you build a native binary, or run tests in native mode, test resources should be available

  • easy to use: the Micronaut build plugins for Gradle and Maven should handle the complexity of figuring out the dependencies for you

  • extensible: you can implement your own test resources, in case the built-in ones do not cover your use case

  • technology agnostic: while lots of test resources use Testcontainers under the hood, you can use any other technology to create resources

14. Next Steps

Read more about transactions in Micronaut Data and Oracle priority transactions.

15. License

All guides are released with an Apache License 2.0 for the code and a Creative Commons Attribution 4.0 license for the writing and media (images).