Table of Contents

Oracle Sessionless Transactions with Micronaut Data and Kafka

Learn how to use Oracle sessionless transactions with Micronaut Data and Kafka.

Authors: Milenko Supic

Micronaut Version: 5.2.3

1. Getting Started

In this guide, we will create a Micronaut application written in Groovy.

In this guide you will build a flight-booking workflow with an HTTP controller and two Kafka consumers. The controller reserves a seat in Oracle Database, suspends its transaction, and publishes a payment request. The first consumer processes a simulated payment. The second resumes the original Oracle Database transaction and confirms the booking, or rolls back the reservation if payment is declined.

The controller and consumers exchange the encoded global transaction identifier (GTRID), rather than a JDBC connection. The payment consumer never resumes the transaction and does not need to read the uncommitted booking.

2. What you will need

To complete this guide, you will need the following:

  • Some time on your hands

  • A decent text editor or IDE (e.g. IntelliJ IDEA)

  • JDK 21 or greater installed with JAVA_HOME configured appropriately

  • Docker, for Oracle Database Free and Kafka through Micronaut Test Resources.

  • Micronaut Data 5.2.0 or later.

  • Oracle AI Database 26ai (version 23.6 or later) and a recent Oracle JDBC Thin driver exposing startTransaction, suspendTransactionImmediately, and resumeTransaction.

Read Developing Applications with Sessionless Transactions in the Oracle AI Database Development Guide for how the feature works, and Sessionless Transactions in the Oracle AI Database JDBC Developer’s Guide for database and driver requirements.

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. The Booking Flow

The following diagram illustrates the booking flow:

Booking flow: the controller reserves the seat and suspends the Oracle Database transaction

  1. Reserve the seat — BookingController

    Receives POST /bookings, reserves the seat, and saves a pending booking. It suspends the Oracle Database transaction, and then publishes a PaymentRequested message containing the booking ID and GTRID to the payment-requests Kafka topic.

  2. Process payment — PaymentConsumer

    Consumes messages from the payment-requests Kafka topic and processes payment while the reservation transaction remains suspended. It publishes a PaymentCompleted message containing the payment outcome, booking ID, and the same GTRID to the payment-results Kafka topic.

  3. Complete the booking — CompletionConsumer

    Consumes messages from the payment-results Kafka topic and resumes the original Oracle Database transaction using the GTRID. If payment is approved, it confirms the booking and commits. If payment is declined, it rolls back the reservation.

The sample runs the controller and both consumers in one application. The consumers have separate consumer groups. They can be deployed in separate applications using the same event contracts. Reservation and completion must connect to the same Oracle Database instance and use the same application database user. No connection is held by the application while payment runs, but the suspended transaction continues to hold its database locks.

This is an Oracle Database transaction spanning workflow steps. Kafka publishing, Kafka offset commits, and the external payment are separate operations. The example demonstrates propagation and completion; it does not provide atomic delivery or exactly-once booking across these systems.

5. Writing the Application

Create an application using the Micronaut Command Line Interface or with Micronaut Launch.

mn create-app example.micronaut.micronautguide \
    --features=data-jdbc,jdbc-hikari,oracle,liquibase,kafka,serialization-jackson,validation,awaitility \
    --build=gradle \
    --lang=groovy \
    --test=spock
If you don’t specify the --build argument, Gradle with the Kotlin DSL is used as the build tool.
If you don’t specify the --lang argument, Java is used as the language.
If you don’t specify the --test argument, JUnit is used for Java and Kotlin, and Spock is used for Groovy.

The previous command creates a Micronaut application with the default package example.micronaut in a directory named micronautguide.

If you use Micronaut Launch, select Micronaut Application as application type and add data-jdbc, jdbc-hikari, oracle, liquibase, kafka, serialization-jackson, validation, and awaitility features.

If you have an existing Micronaut application and want to add the functionality described here, you can view the dependency and configuration changes from the specified features, and apply those changes to your application.

Diff

Setup IntelliJ IDEA to develop Micronaut Applications.

5.1. Oracle Driver

Add also the Oracle Driver

build.gradle
runtimeOnly("com.oracle.database.jdbc:ojdbc11")

6. Configure Oracle Database and Kafka

Configure the default datasource for Oracle Database:

src/main/resources/application.properties
datasources.default.driver-class-name=oracle.jdbc.OracleDriver
datasources.default.db-type=oracle
datasources.default.dialect=ORACLE
datasources.default.schema-generate=NONE

6.1. Database Migration with Liquibase

We need a way to create the database schema. For that, we use Micronaut integration with Liquibase.

Add the following snippet to include the necessary dependencies:

build.gradle
implementation("io.micronaut.liquibase:micronaut-liquibase")

Configure the database migrations directory for Liquibase in application.properties.

src/main/resources/application.properties
liquibase.datasources.default.change-log=classpath\:db/liquibase-changelog.xml

Create the following files with the database schema creation:

src/main/resources/db/liquibase-changelog.xml
<?xml version="1.0" encoding="UTF-8"?>
<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
                  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
                  xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog https://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.20.xsd">
    <changeSet id="1" author="micronaut">
        <sqlFile path="booking.sql" relativeToChangelogFile="true"/>
    </changeSet>
    <changeSet id="2" author="micronaut">
        <sqlFile path="seed.sql" relativeToChangelogFile="true"/>
    </changeSet>
</databaseChangeLog>

Create the schema that Liquibase applies. Schema generation is disabled so that application restarts preserve committed bookings:

src/main/resources/db/booking.sql
CREATE TABLE flight_seat (
    id VARCHAR2(40) PRIMARY KEY,
    status VARCHAR2(20) NOT NULL CHECK (status IN ('AVAILABLE', 'HELD', 'BOOKED'))
);
CREATE SEQUENCE booking_seq START WITH 1 INCREMENT BY 1;
CREATE TABLE booking (
    id NUMBER(19) PRIMARY KEY,
    seat_id VARCHAR2(40) NOT NULL REFERENCES flight_seat(id),
    amount NUMBER(10, 2) NOT NULL,
    status VARCHAR2(20) NOT NULL CHECK (status IN ('PENDING', 'CONFIRMED')),
    payment_reference VARCHAR2(80) NOT NULL
);

Create the seed data. The changelog inserts two available seats when it initializes a new database:

src/main/resources/db/seed.sql
INSERT INTO flight_seat (id, status) VALUES ('MN123-12A', 'AVAILABLE');
INSERT INTO flight_seat (id, status) VALUES ('MN123-12B', 'AVAILABLE');

6.2. Kafka Configuration

Configure the Kafka producer and consumers:

src/main/resources/application.properties
kafka.consumers.default.auto.offset.reset=earliest
kafka.consumers.default.enable.auto.commit=false
kafka.producers.default.enable.idempotence=true

The consumers start from the earliest available offset for a new group. Automatic offset commits are disabled; each listener uses SYNC_PER_RECORD. The producer requests acknowledgments from all in-sync replicas and enables Kafka producer idempotence. That setting covers producer retries; it does not deduplicate repeated business operations.

6.3. Test Resources Configuration

Configure Micronaut Test Resources to start an Oracle Database Free image that supports sessionless transactions in the tests:

src/test/resources/application-test.properties
test-resources.containers.oracle.image-name=gvenzl/oracle-free
test-resources.containers.oracle.image-tag=23.26.3-slim-faststart
test-resources.containers.oracle.startup-timeout=600s

Micronaut Test Resources supplies the datasource URL, username, password, and kafka.bootstrap.servers when they are absent. The pinned Oracle Database Free image supports sessionless transactions. For your own services, set DATASOURCES_DEFAULT_URL, DATASOURCES_DEFAULT_USERNAME, DATASOURCES_DEFAULT_PASSWORD, and KAFKA_BOOTSTRAP_SERVERS; do not put credentials in source control.

Create the two topics before sending messages when using a broker without topic auto-creation:

for topic in payment-requests payment-results; do
  kafka-topics.sh --bootstrap-server localhost:9092 --create --if-not-exists \
    --topic "$topic" --partitions 1 --replication-factor 1
done

One replica is suitable for the local demonstration. Configure replication and minimum in-sync replicas for your deployment.

7. Application Packages

The source code is organized under example.micronaut:

  • controller: HTTP booking endpoints.

  • dto: HTTP request and response models and Kafka event payloads.

  • entity: persistence entities and status enums.

  • repository: Micronaut Data repositories.

  • service: transactional booking operations and the demonstration payment gateway.

  • kafka: the producer client and the two consumers.

The end-to-end tests exercise the booking controller.

8. Entities and Repositories

Create the FlightSeat entity, which maps the flight_seat table:

src/main/groovy/example/micronaut/entity/FlightSeat.groovy
package example.micronaut.entity

import groovy.transform.CompileStatic
import io.micronaut.data.annotation.Id
import io.micronaut.data.annotation.MappedEntity

@CompileStatic
@MappedEntity("flight_seat") (1)
class FlightSeat {
    @Id (2)
    String id
    SeatStatus status

    FlightSeat(String id, SeatStatus status) {
        this.id = id
        this.status = status
    }
}
1 Annotate the class with @MappedEntity to map the class to the table defined in the schema.
2 Specifies the ID of an entity

Create the SeatStatus enum:

src/main/groovy/example/micronaut/entity/SeatStatus.groovy
package example.micronaut.entity

import groovy.transform.CompileStatic

import io.micronaut.serde.annotation.Serdeable

@CompileStatic
@Serdeable (1)
enum SeatStatus {
    AVAILABLE,
    HELD,
    BOOKED
}
1 Declare the @Serdeable annotation at the type level in your source code to allow the type to be serialized or deserialized.

SeatStatus represents an available seat, a held reservation, and a completed booking. Micronaut Data stores the enum names in the existing VARCHAR2 status column. HTTP responses expose the same names through FlightSeatView, so the enum has @Serdeable while the entity does not.

Create the Booking entity, which maps the booking table:

src/main/groovy/example/micronaut/entity/Booking.groovy
package example.micronaut.entity

import groovy.transform.CompileStatic
import io.micronaut.data.annotation.Id
import io.micronaut.data.annotation.MappedEntity
import io.micronaut.data.annotation.GeneratedValue

@CompileStatic
@MappedEntity("booking") (1)
class Booking {
    @Id (2)
    @GeneratedValue (3)
    Long id
    String seatId
    BigDecimal amount
    BookingStatus status
    String paymentReference

    Booking(String seatId, BigDecimal amount, BookingStatus status, String paymentReference) {
        this.seatId = seatId
        this.amount = amount
        this.status = status
        this.paymentReference = paymentReference
    }
}
1 Annotate the class with @MappedEntity to map the class to the table defined in the schema.
2 Specifies the ID of an entity
3 Specifies that the property value is generated by the database and not included in inserts

Create the BookingStatus enum:

src/main/groovy/example/micronaut/entity/BookingStatus.groovy
package example.micronaut.entity

import groovy.transform.CompileStatic

import io.micronaut.serde.annotation.Serdeable

@CompileStatic
@Serdeable (1)
enum BookingStatus {
    PENDING,
    CONFIRMED
}
1 Declare the @Serdeable annotation at the type level in your source code to allow the type to be serialized or deserialized.

BookingStatus represents the pending and confirmed states. Micronaut Data stores the enum names in the existing VARCHAR2 status column, and the HTTP response serializes them as PENDING or CONFIRMED. The enum is serializable because it appears in BookingView; the persistence entity still has no serialization annotation.

The HTTP endpoints return DTOs instead of persistence entities. Create FlightSeatView to represent a seat:

src/main/groovy/example/micronaut/dto/FlightSeatView.groovy
package example.micronaut.dto

import groovy.transform.Canonical
import groovy.transform.CompileStatic
import io.micronaut.serde.annotation.Serdeable
import example.micronaut.entity.SeatStatus

@Canonical
@CompileStatic
@Serdeable (1)
class FlightSeatView {
    String id
    SeatStatus status
}
1 Declare the @Serdeable annotation at the type level in your source code to allow the type to be serialized or deserialized.

Create BookingView to represent a booking:

src/main/groovy/example/micronaut/dto/BookingView.groovy
package example.micronaut.dto

import groovy.transform.Canonical
import groovy.transform.CompileStatic
import io.micronaut.serde.annotation.Serdeable
import example.micronaut.entity.BookingStatus

@Canonical
@CompileStatic
@Serdeable (1)
class BookingView {
    Long id
    String seatId
    BigDecimal amount
    BookingStatus status
    String paymentReference
}
1 Declare the @Serdeable annotation at the type level in your source code to allow the type to be serialized or deserialized.

Create a repository interface for seats. Micronaut Data implements the interface at compilation time:

src/main/groovy/example/micronaut/repository/FlightSeatRepository.groovy
package example.micronaut.repository

import groovy.transform.CompileStatic

import example.micronaut.dto.FlightSeatView
import example.micronaut.entity.FlightSeat
import example.micronaut.entity.SeatStatus

import io.micronaut.data.jdbc.annotation.JdbcRepository
import io.micronaut.data.model.query.builder.sql.Dialect
import io.micronaut.data.repository.CrudRepository
import java.util.Optional

@CompileStatic
@JdbcRepository(dialect = Dialect.ORACLE) (1)
interface FlightSeatRepository extends CrudRepository<FlightSeat, String> { (2)
    Optional<FlightSeat> findByIdAndStatusForUpdate(String id, SeatStatus status)
    Optional<FlightSeat> findByIdAndStatus(String id, SeatStatus status)
    Optional<FlightSeatView> queryById(String id) (3)
}
1 @JdbcRepository with a specific dialect.
2 By extending CrudRepository you enable automatic generation of CRUD (Create, Read, Update, Delete) operations.
3 Micronaut Data supports reflection-free Data Transfer Object (DTO) projections if the return type is annotated with @Introspected.

The derived findByIdAndStatusForUpdate method lets Micronaut Data generate a locking read for the Oracle Database dialect that returns the seat only when it is AVAILABLE. The service persists the new status with the standard update method while holding this lock. During completion, findByIdAndStatus retrieves only a seat whose status is HELD. Oracle Database holds the lock until the original transaction commits or rolls back, including while it is suspended. Another reservation for the same seat can wait on that lock. A normal read still sees the previously committed AVAILABLE value; it cannot tell you whether a suspended reservation holds the seat.

Create a repository interface for bookings:

src/main/groovy/example/micronaut/repository/BookingRepository.groovy
package example.micronaut.repository

import groovy.transform.CompileStatic

import example.micronaut.dto.BookingView
import example.micronaut.entity.Booking
import example.micronaut.entity.BookingStatus

import io.micronaut.data.jdbc.annotation.JdbcRepository
import io.micronaut.data.model.query.builder.sql.Dialect
import io.micronaut.data.repository.CrudRepository
import java.util.Optional

@CompileStatic
@JdbcRepository(dialect = Dialect.ORACLE) (1)
interface BookingRepository extends CrudRepository<Booking, Long> { (2)
    Optional<Booking> findByIdAndStatus(Long id, BookingStatus status)
    Optional<BookingView> queryById(Long id) (3)
}
1 @JdbcRepository with a specific dialect.
2 By extending CrudRepository you enable automatic generation of CRUD (Create, Read, Update, Delete) operations.
3 Micronaut Data supports reflection-free Data Transfer Object (DTO) projections if the return type is annotated with @Introspected.

The Java FlightSeat and Booking entities are records, which Micronaut Data JDBC supports; the Groovy, Kotlin, and Python variants use mutable entities. The Java service creates new record values with the updated status. None of the persistence entities has serialization annotations. The booking ID uses @GeneratedValue with its default settings. For Oracle Database, Micronaut Data selects sequence generation and derives the sequence name from the table name plus _seq, so Liquibase creates booking_seq for the booking table. A new booking has no assigned ID; the inherited save method inserts it and returns the entity with its generated ID. Sequence values can have gaps after rollback. During completion, the derived findByIdAndStatus method retrieves only a booking that is still PENDING. Changes to seats and bookings use the inherited update method. The queryById methods return the FlightSeatView and BookingView DTOs directly, so Micronaut Data selects only their properties and the controller does not map entities. Both repositories let Micronaut Data generate their SQL.

9. Request and Message Contracts

Create BookingRequest to bind the body of the HTTP booking request:

src/main/groovy/example/micronaut/dto/BookingRequest.groovy
package example.micronaut.dto

import groovy.transform.Canonical
import groovy.transform.CompileStatic
import io.micronaut.serde.annotation.Serdeable
import jakarta.validation.constraints.NotBlank
import jakarta.validation.constraints.Pattern

@Canonical
@CompileStatic
@Serdeable (1)
class BookingRequest {
    @NotBlank (2)
    String seatId
    @NotBlank
    @Pattern(regexp = "demo-approve|demo-decline")
    String paymentToken
}
1 Declare the @Serdeable annotation at the type level in your source code to allow the type to be serialized or deserialized.
2 Use jakarta.validation.constraints Constraints to ensure the data matches your expectations.

Create the PaymentRequested event, which the controller publishes to request a payment:

src/main/groovy/example/micronaut/dto/PaymentRequested.groovy
package example.micronaut.dto

import groovy.transform.Canonical
import groovy.transform.CompileStatic
import io.micronaut.serde.annotation.Serdeable

@Canonical
@CompileStatic
@Serdeable (1)
class PaymentRequested {
    Long bookingId
    String transactionId
    BigDecimal amount
    String paymentToken
}
1 Declare the @Serdeable annotation at the type level in your source code to allow the type to be serialized or deserialized.

Create the PaymentCompleted event, which carries the payment outcome:

src/main/groovy/example/micronaut/dto/PaymentCompleted.groovy
package example.micronaut.dto

import groovy.transform.Canonical
import groovy.transform.CompileStatic
import io.micronaut.serde.annotation.Serdeable

@Canonical
@CompileStatic
@Serdeable (1)
class PaymentCompleted {
    Long bookingId
    String transactionId
    boolean approved
    String paymentReference
}
1 Declare the @Serdeable annotation at the type level in your source code to allow the type to be serialized or deserialized.

The controller passes only seatId to the booking service and uses the payment token from the HTTP request when constructing the payment message. Kafka messages contain only the dedicated PaymentRequested and PaymentCompleted DTOs. Persistence entities are never sent to Kafka. All events use the database-generated booking ID as their Kafka key. The controller gets this ID from the saved booking before constructing the payment request. The payment request carries the server-selected price because the pending booking is not visible outside its suspended transaction.

Micronaut’s propagation API returns the Oracle Database GTRID as a string encoded by OracleSessionlessTransactionIdCodec. Include this string in the transactionId field of the Kafka messages so the final consumer can resume the transaction. You can customize the codec, but the receiving application must be able to decode the transaction IDs sent by the producing application.

Treat the encoded GTRID as sensitive transaction-control data. Restrict access to these topics, use authenticated, encrypted connections outside local development, and avoid logging it. Do not send card details in Kafka messages; the sample uses only demonstration tokens.

10. Transactional Booking Service

Create an exception to signal a declined payment:

src/main/groovy/example/micronaut/service/PaymentDeclinedException.groovy
package example.micronaut.service
import groovy.transform.CompileStatic


@CompileStatic
class PaymentDeclinedException extends RuntimeException {
    PaymentDeclinedException() {
        super("Payment declined; roll back the seat reservation")
    }
}

Create BookingService. It reserves a seat in a suspended transaction and later resumes that transaction to confirm or roll back the booking:

src/main/groovy/example/micronaut/service/BookingService.groovy
package example.micronaut.service

import groovy.transform.CompileStatic

import example.micronaut.dto.PaymentCompleted
import example.micronaut.entity.Booking
import example.micronaut.entity.BookingStatus
import example.micronaut.entity.FlightSeat
import example.micronaut.entity.SeatStatus
import example.micronaut.repository.BookingRepository
import example.micronaut.repository.FlightSeatRepository

import io.micronaut.transaction.annotation.OracleTransactional
import jakarta.inject.Singleton
import java.math.BigDecimal

import static io.micronaut.transaction.annotation.OracleTransactional.Sessionless.REQUIRES_SUSPENDED
import static io.micronaut.transaction.annotation.OracleTransactional.Sessionless.SUSPEND

@CompileStatic
@Singleton (9)
class BookingService {
    private final FlightSeatRepository flightSeatRepository
    private final BookingRepository bookingRepository

    BookingService(FlightSeatRepository flightSeatRepository, BookingRepository bookingRepository) {
        this.flightSeatRepository = flightSeatRepository
        this.bookingRepository = bookingRepository
    }

    @OracleTransactional(sessionless = SUSPEND, timeout = 300) (1)
    Booking reserve(String seatId) {
        FlightSeat seat = flightSeatRepository.findByIdAndStatusForUpdate(seatId, SeatStatus.AVAILABLE).orElseThrow() (2)
        seat.setStatus(SeatStatus.HELD)
        flightSeatRepository.update(seat)
        BigDecimal amount = new BigDecimal("199.00")
        return bookingRepository.save(new Booking(seatId, amount, BookingStatus.PENDING, "NONE")) (3)
    }

    @OracleTransactional(sessionless = REQUIRES_SUSPENDED) (4)
    void complete(PaymentCompleted result) {
        Booking booking = bookingRepository.findByIdAndStatus(result.bookingId, BookingStatus.PENDING).orElseThrow() (5)
        if (!result.approved) {
            throw new PaymentDeclinedException() (6)
        }
        FlightSeat seat = flightSeatRepository.findByIdAndStatus(booking.getSeatId(), SeatStatus.HELD).orElseThrow() (7)
        seat.setStatus(SeatStatus.BOOKED)
        flightSeatRepository.update(seat)
        booking.setStatus(BookingStatus.CONFIRMED)
        booking.setPaymentReference(result.paymentReference)
        bookingRepository.update(booking)
    } (8)
}
1 Start an Oracle Database sessionless transaction with a five-minute suspension timeout. Oracle Database cancels a transaction that remains suspended past its timeout.
2 Lock the seat with the generated findByIdAndStatusForUpdate query, which returns the seat only when it is AVAILABLE. Change its status to HELD and call update; create the pending booking with save. A competing reservation waits for the lock, and Oracle Database then re-evaluates the status condition against the latest committed row, preventing two bookings of the same seat. If no available seat is found, orElseThrow throws an exception and rolls back the transaction.
3 save returns the booking with its database-generated ID. After the intercepted method returns, Micronaut suspends the transaction instead of committing it. The generated ID and price can be published, but the database writes are still uncommitted.
4 Resume the transaction whose GTRID is in the active propagation scope.
5 This derived finder runs inside the resumed transaction and returns only the pending booking written by the controller.
6 A runtime exception rolls back the original transaction, undoing both the seat reservation and the pending booking.
7 Find the seat only when its status is HELD; if either derived finder returns no row, the operation fails and the suspended transaction is not committed.
8 Update the seat and persist the confirmed booking through the standard repository update methods. A normal return commits both changes together. Micronaut clears the GTRID from the propagation state after completion.
9 Use jakarta.inject.Singleton to designate a class as a singleton.

These methods are top-level transaction boundaries on an injected bean. Do not call them inside another active transaction or through self-invocation. The sessionless modes require the default REQUIRED propagation and cannot be combined with @OracleTransactional.Recoverable.

11. Kafka Producer

Create an interface to send messages to Kafka. The Micronaut framework implements the interface at compilation time:

src/main/groovy/example/micronaut/kafka/BookingClient.groovy
package example.micronaut.kafka

import groovy.transform.CompileStatic

import example.micronaut.dto.PaymentCompleted
import example.micronaut.dto.PaymentRequested

import io.micronaut.configuration.kafka.annotation.KafkaClient
import io.micronaut.configuration.kafka.annotation.KafkaKey
import io.micronaut.configuration.kafka.annotation.Topic
import org.apache.kafka.clients.producer.RecordMetadata

@CompileStatic
@KafkaClient(acks = KafkaClient.Acknowledge.ALL) (1)
interface BookingClient {
    @Topic("payment-requests") (2)
    RecordMetadata requestPayment(@KafkaKey Long bookingId, PaymentRequested request) (3)

    @Topic("payment-results")
    RecordMetadata paymentCompleted(@KafkaKey Long bookingId, PaymentCompleted result)
}
1 The @KafkaClient annotation is used to designate this interface as a client
2 The @Topic annotation indicates which topics the @KafkaClient should be published to.
3 The Kafka key can be specified by providing a parameter annotated with @KafkaKey. If no such parameter is specified the record is sent with a null key.

Returning RecordMetadata makes each producer method synchronous: it waits for a Kafka acknowledgment before returning. The controller waits for its payment request to be acknowledged before returning 202 Accepted. The payment consumer waits for its result publication before its input offset is committed. There is still a crash window between suspending the Oracle Database transaction and receiving a Kafka acknowledgment.

12. Controller: Reserve and Suspend

Create BookingController to reserve seats and to expose bookings and seats over HTTP:

src/main/groovy/example/micronaut/controller/BookingController.groovy
package example.micronaut.controller

import groovy.transform.CompileStatic

import example.micronaut.dto.BookingRequest
import example.micronaut.dto.BookingView
import example.micronaut.dto.FlightSeatView
import example.micronaut.dto.PaymentRequested
import example.micronaut.entity.Booking
import example.micronaut.kafka.BookingClient
import example.micronaut.repository.BookingRepository
import example.micronaut.repository.FlightSeatRepository
import example.micronaut.service.BookingService

import io.micronaut.http.HttpResponse
import io.micronaut.http.annotation.Body
import io.micronaut.http.annotation.Controller
import io.micronaut.http.annotation.Get
import io.micronaut.http.annotation.Post
import io.micronaut.scheduling.TaskExecutors
import io.micronaut.scheduling.annotation.ExecuteOn
import io.micronaut.serde.annotation.Serdeable
import io.micronaut.transaction.jdbc.oracle.OracleSessionlessTransactionPropagationOperations
import jakarta.validation.Valid
import java.util.Optional

@CompileStatic
@Controller("/bookings") (1)
@ExecuteOn(TaskExecutors.BLOCKING) (2)
class BookingController {
    private final BookingClient bookingClient
    private final BookingService bookingService
    private final OracleSessionlessTransactionPropagationOperations propagation
    private final BookingRepository bookingRepository
    private final FlightSeatRepository flightSeatRepository

    BookingController(BookingClient bookingClient,
                             BookingService bookingService,
                             OracleSessionlessTransactionPropagationOperations propagation,
                             BookingRepository bookingRepository,
                             FlightSeatRepository flightSeatRepository) {
        this.bookingClient = bookingClient
        this.bookingService = bookingService
        this.propagation = propagation
        this.bookingRepository = bookingRepository
        this.flightSeatRepository = flightSeatRepository
    }

    @Serdeable (10)
    @groovy.transform.Canonical
    static class AcceptedBooking {
        Long bookingId
    }

    @Post (3)
    HttpResponse<AcceptedBooking> book(@Body @Valid BookingRequest request) {
        PaymentRequested payment = propagation.withPropagation({ -> (4)
            Booking booking = bookingService.reserve(request.seatId) (5)
            String transactionId = propagation.currentTransactionId().orElseThrow() (6)
            return new PaymentRequested(booking.getId(), transactionId, booking.getAmount(), request.paymentToken)
        })
        bookingClient.requestPayment(payment.bookingId, payment) (7)
        return HttpResponse.accepted().body(new AcceptedBooking(payment.bookingId))
    }

    @Get("/{id}") (8)
    Optional<BookingView> find(Long id) {
        return bookingRepository.queryById(id)
    }

    @Get("/seats/{id}") (9)
    Optional<FlightSeatView> seat(String id) {
        return flightSeatRepository.queryById(id)
    }
}
1 The class is defined as a controller with the @Controller annotation mapped to the path /bookings.
2 It is critical that any blocking I/O operations (such as fetching the data from the database) are offloaded to a separate thread pool that does not block the Event loop.
3 The @Post annotation maps the book method to an HTTP POST request on /bookings.
4 Create an empty lexical Oracle Database propagation scope for the incoming HTTP booking request.
5 Call the injected transactional service. It saves the booking and returns the entity with its generated ID and price. Its interceptor suspends the Oracle Database transaction before returning to the controller.
6 Obtain the encoded GTRID after suspension, while the propagation scope is still active.
7 Publish the payment request after the transaction has been suspended. The event carries the GTRID even though the lexical scope has ended.
8 The @Get annotation maps the find method to an HTTP GET request on /bookings/{id}.
9 The @Get annotation maps the seat method to an HTTP GET request on /bookings/seats/{id}.
10 Declare the @Serdeable annotation at the type level in your source code to allow the type to be serialized or deserialized.

The controller returns 202 Accepted with a booking ID after Kafka acknowledges the payment request. Reservation has already run and the Oracle Database transaction is suspended, but payment and booking completion are asynchronous. Blocking Kafka sends and JDBC work run on the blocking executor. The controller is an HTTP entry point; it does not listen to a Kafka topic.

Do not publish the payment request from inside the SUSPEND method: a consumer could receive it before Oracle Database has suspended the transaction. A lexical propagation scope is not transported automatically across Kafka or asynchronous threads.

13. First Consumer: Process Payment

Create a demonstration payment gateway:

src/main/groovy/example/micronaut/service/DemoPaymentGateway.groovy
package example.micronaut.service

import groovy.transform.CompileStatic

import example.micronaut.dto.PaymentRequested

import jakarta.inject.Singleton
import java.util.Map
import java.util.concurrent.ConcurrentHashMap

@CompileStatic
@Singleton (1)
class DemoPaymentGateway {
    private final Map<Long, Decision> decisions = new ConcurrentHashMap<>()

    @groovy.transform.Canonical
    static class Decision {
        boolean approved
        String reference
    }

    Decision pay(PaymentRequested request) {
        return decisions.computeIfAbsent(request.bookingId, { Long id ->
            boolean approved = "demo-approve".equals(request.paymentToken)
            return new Decision(approved, approved ? "demo-" + id : "DECLINED")
        })
    }
}
1 Use jakarta.inject.Singleton to designate a class as a singleton.

The gateway simulates an approval for demo-approve and a decline for demo-decline. Its in-memory map makes repeated payments with the same booking ID return the same result during one process lifetime. Replace this with a payment provider that accepts a durable idempotency key; the demonstration map does not survive restarts.

Create PaymentConsumer. It listens to the payment-requests topic, processes the payment, and publishes the result:

src/main/groovy/example/micronaut/kafka/PaymentConsumer.groovy
package example.micronaut.kafka

import groovy.transform.CompileStatic

import example.micronaut.dto.PaymentCompleted
import example.micronaut.dto.PaymentRequested
import example.micronaut.service.DemoPaymentGateway

import io.micronaut.configuration.kafka.annotation.KafkaListener
import io.micronaut.configuration.kafka.annotation.OffsetStrategy
import io.micronaut.configuration.kafka.annotation.Topic

@CompileStatic
@KafkaListener(groupId = "payment-processing", offsetStrategy = OffsetStrategy.SYNC_PER_RECORD) (1)
class PaymentConsumer {
    private final DemoPaymentGateway demoPaymentGateway
    private final BookingClient bookingClient

    PaymentConsumer(DemoPaymentGateway demoPaymentGateway, BookingClient bookingClient) {
        this.demoPaymentGateway = demoPaymentGateway
        this.bookingClient = bookingClient
    }

    @Topic("payment-requests") (2)
    void pay(PaymentRequested request) {
        DemoPaymentGateway.Decision decision = demoPaymentGateway.pay(request) (3)
        bookingClient.paymentCompleted(request.bookingId, new PaymentCompleted(
            request.bookingId, request.transactionId, decision.approved, decision.reference)) (4)
    }
}
1 Mark the class with @KafkaListener to register it as a consumer. Its group ID identifies the consumer group, and its offset strategy controls when consumed offsets are committed.
2 The @Topic annotation is again used to indicate which topic(s) to subscribe to.
3 Process payment without entering an Oracle Database transaction or querying the pending booking.
4 Publish the result with the exact GTRID received in the payment request.

14. Second Consumer: Resume and Complete

Create CompletionConsumer. It listens to the payment-results topic and completes the booking:

src/main/groovy/example/micronaut/kafka/CompletionConsumer.groovy
package example.micronaut.kafka

import groovy.transform.CompileStatic

import example.micronaut.dto.PaymentCompleted
import example.micronaut.entity.BookingStatus
import example.micronaut.repository.BookingRepository
import example.micronaut.service.BookingService
import example.micronaut.service.PaymentDeclinedException

import io.micronaut.configuration.kafka.annotation.KafkaListener
import io.micronaut.configuration.kafka.annotation.OffsetStrategy
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.transaction.jdbc.oracle.OracleSessionlessTransactionPropagationOperations

@CompileStatic
@KafkaListener(groupId = "booking-completion", offsetStrategy = OffsetStrategy.SYNC_PER_RECORD) (1)
class CompletionConsumer {
    private final BookingService bookingService
    private final BookingRepository bookingRepository
    private final OracleSessionlessTransactionPropagationOperations propagation

    CompletionConsumer(BookingService bookingService, BookingRepository bookingRepository,
                              OracleSessionlessTransactionPropagationOperations propagation) {
        this.bookingService = bookingService
        this.bookingRepository = bookingRepository
        this.propagation = propagation
    }

    @Topic("payment-results") (2)
    void complete(PaymentCompleted result) {
        if (bookingRepository.findById(result.bookingId).filter({ b -> b.status == BookingStatus.CONFIRMED }).isPresent()) {
            return (3)
        }
        try {
            propagation.withPropagation(result.transactionId, { -> (4)
                bookingService.complete(result) (5)
                return null
            })
        } catch (PaymentDeclinedException expected) { (6)
            // The transactional interceptor has already rolled back the reservation.
        }
    }
}
1 Mark the class with @KafkaListener to register it as a consumer. Its group ID identifies the consumer group, and its offset strategy controls when consumed offsets are committed.
2 The @Topic annotation is again used to indicate which topic(s) to subscribe to.
3 Recognize an already committed confirmation before attempting to resume its completed GTRID. This handles sequential duplicates of a successful result; it is not a complete deduplication mechanism.
4 Import the encoded GTRID into a fresh propagation scope on the consumer thread.
5 Resume the original transaction and complete the booking through the injected service.
6 Catch the expected decline only outside the transactional service, after rollback. Catching it inside that method would allow the reservation to commit.

A declined booking leaves no booking row: the original insert was rolled back. An absent booking can also mean it is pending, has expired, or was never started. A production workflow needs a separate durable status record to distinguish those outcomes and recognize duplicate declines.

15. End-to-End Tests

Create an end-to-end test that books seats through the HTTP API. Liquibase seeds the two demonstration seats only once per database, and each test books one of them. Test Resources can reuse the same database, for example when native tests run after the JVM tests, so each test cleans up after itself through the repositories:

src/test/groovy/example/micronaut/controller/KafkaBookingSpec.groovy
package example.micronaut.controller

import example.micronaut.dto.BookingRequest
import example.micronaut.dto.BookingView
import example.micronaut.dto.FlightSeatView
import example.micronaut.entity.BookingStatus
import example.micronaut.entity.FlightSeat
import example.micronaut.entity.SeatStatus
import example.micronaut.repository.BookingRepository
import example.micronaut.repository.FlightSeatRepository
import io.micronaut.context.annotation.Property
import io.micronaut.http.HttpRequest
import io.micronaut.http.HttpStatus
import io.micronaut.http.client.HttpClient
import io.micronaut.http.client.annotation.Client
import io.micronaut.http.client.exceptions.HttpClientResponseException
import io.micronaut.test.extensions.spock.annotation.MicronautTest
import jakarta.inject.Inject
import spock.lang.Specification
import spock.lang.Timeout

import java.time.Duration
import static org.awaitility.Awaitility.await

@MicronautTest(transactional = false) (1)
@Property(name = "micronaut.test.resources.scope", value = "booking-controller-e2e") (2)
@Property(name = "micronaut.http.client.read-timeout", value = "30s")
@Timeout(120)
class KafkaBookingSpec extends Specification {
    @Inject
    @Client("/")
    HttpClient httpClient

    @Inject
    BookingRepository bookingRepository

    @Inject
    FlightSeatRepository flightSeatRepository

    void cleanup() { (3)
        bookingRepository.deleteAll()
        flightSeatRepository.update(new FlightSeat("MN123-12A", SeatStatus.AVAILABLE))
        flightSeatRepository.update(new FlightSeat("MN123-12B", SeatStatus.AVAILABLE))
    }

    void "approved payment confirms the booking"() {
        given:
        String seatId = "MN123-12A"
        assert findSeat(seatId).status == SeatStatus.AVAILABLE

        when:
        Long bookingId = book(seatId, "demo-approve")

        then:
        awaitConfirmed(bookingId, seatId)
    }

    void "declined payment rolls back and allows the same seat to be booked again"() {
        given:
        String seatId = "MN123-12B"
        assert findSeat(seatId).status == SeatStatus.AVAILABLE

        when:
        Long declinedBookingId = book(seatId, "demo-decline") (4)
        Long approvedBookingId = book(seatId, "demo-approve") (5)

        then:
        declinedBookingId != approvedBookingId
        awaitConfirmed(approvedBookingId, seatId) (6)
        findBooking(declinedBookingId).empty (7)
    }

    private Long book(String seatId, String paymentToken) {
        def response = httpClient.toBlocking().exchange(
            HttpRequest.POST("/bookings", new BookingRequest(seatId, paymentToken)),
            BookingController.AcceptedBooking)
        assert response.status == HttpStatus.ACCEPTED
        Long bookingId = response.body().bookingId
        assert bookingId != null && bookingId > 0
        bookingId
    }

    private boolean awaitConfirmed(Long bookingId, String seatId) {
        await().atMost(Duration.ofSeconds(60)).untilAsserted {
            Optional<BookingView> booking = findBooking(bookingId)
            assert booking.present : "Booking has not committed yet"
            BookingView view = booking.orElseThrow()
            assert view.id == bookingId
            assert view.seatId == seatId
            assert view.status == BookingStatus.CONFIRMED
            assert view.paymentReference == "demo-" + bookingId
            assert findSeat(seatId).status == SeatStatus.BOOKED
        }
        true
    }

    private Optional<BookingView> findBooking(Long bookingId) {
        try {
            Optional.of(httpClient.toBlocking().retrieve("/bookings/" + bookingId, BookingView))
        } catch (HttpClientResponseException e) {
            if (e.status != HttpStatus.NOT_FOUND) throw e
            Optional.empty()
        }
    }

    private FlightSeatView findSeat(String seatId) {
        httpClient.toBlocking().retrieve("/bookings/seats/" + seatId, FlightSeatView)
    }
}
1 Configure the test with @MicronautTest so Micronaut initializes the application context and embedded server. Tests are transactional by default, and each test transaction rolls back when the test finishes. Set transactional to false when the test must observe committed database changes.
2 Annotate the class with @Property to supply configuration to the test.
3 After each test, delete the bookings with BookingRepository and make both demonstration seats AVAILABLE again with FlightSeatRepository.
4 Submit a declined booking for the seeded MN123-12B seat.
5 Submit an approved booking for the exact same seat. This request must acquire the lock held by the first reservation, so it can return only after that lock is released.
6 Poll until the replacement booking is CONFIRMED and its seat is BOOKED.
7 Verify the declined booking lookup returns 404, showing that its pending insert was rolled back.

The test submits booking requests through the HTTP API and polls the booking and seat endpoints to verify the outcome. The approval scenario books the seeded MN123-12A seat, then polls until the booking is CONFIRMED with a payment reference and the seat is BOOKED.

An absent booking or an AVAILABLE seat alone would be a false positive: those are also visible while a reservation is suspended. Successfully booking the same seat again and confirming that the declined booking remains absent verify rollback through the public API. The tests use a 30-second HTTP request timeout and a 60-second polling deadline, both shorter than the five-minute Oracle Database suspension timeout, so transaction expiry cannot make this scenario pass.

The test has no surrounding transaction and uses a dedicated Micronaut Test Resources scope, with its own Oracle Database and Kafka containers. Docker is required; the build stops the disposable Test Resources services after the tests.

16. Testing the Application

To run the tests:

./gradlew test

Then open build/reports/tests/test/index.html in a browser to see the results.

17. Try a Booking

Start the application with ./gradlew run.

Test Resources starts local Oracle Database and Kafka containers if external connection properties are absent.

Liquibase inserts MN123-12A and MN123-12B when it initializes a new database. These rows are available for the booking requests below; after a successful booking, that seat remains BOOKED across application restarts.

curl -i -X POST http://localhost:8080/bookings \
  -H 'Content-Type: application/json' \
  -d '{"seatId":"MN123-12A","paymentToken":"demo-approve"}'

Copy bookingId from the response and poll:

curl http://localhost:8080/bookings/BOOKING_ID
curl http://localhost:8080/bookings/seats/MN123-12A

Before completion, the booking lookup returns 404; after completion, it returns a booking with status: CONFIRMED and the seat is BOOKED.

Try MN123-12B with demo-decline. After payment processing and rollback, no booking is visible and the seat remains AVAILABLE. The sample has no separate rejection-status endpoint. Use a different available seat for later successful requests, or recreate the disposable database to reset the demonstration.

18. Cleanup

Stop the application and its Test Resources services.

Use ./gradlew stopTestResourcesService to stop the Gradle Test Resources service.

The containers are disposable; no cloud resources are provisioned by this guide. Remove demonstration topics only from a broker dedicated to this example.

19. Next Steps

Read the Micronaut Data sessionless transaction documentation, the Micronaut Kafka guide, and the Oracle AI Database documentation on sessionless transactions: Developing Applications with Sessionless Transactions and JDBC Sessionless Transactions.

20. 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).