# How to Handle Transactions in Kourier for Atomic Operations

> Learn how to handle transactions in Kourier for atomic operations using txSelect txCommit and txRollback methods for reliable message publishing. Ensure all or no messages are delivered.

- Repository: [Guimauve Digital/kourier](https://github.com/guimauvedigital/kourier)
- Tags: how-to-guide
- Published: 2026-03-02

---

**Kourier provides first-class AMQP transactional support through `txSelect()`, `txCommit()`, and `txRollback()` methods on `AMQPChannel`, enabling atomic message publishing where either all messages are delivered or none are.**

Handling transactions in Kourier for atomic operations ensures that groups of messages are published as a single unit, maintaining data consistency across distributed systems. The `guimauvedigital/kourier` library implements the AMQP 0-9-1 transaction protocol through both low-level channel operations and high-level Kotlin DSL extensions.

## Understanding AMQP Transactions in Kourier

Kourier implements the standard AMQP transactional model that buffers published messages until explicitly committed. When you handle transactions in Kourier for atomic operations, you work with three distinct phases:

1. **Transaction selection** – Switch the channel into transaction mode using `txSelect()`, which sets the internal `isTxMode` flag to `true`.
2. **Message publishing** – All subsequent `basicPublish` calls are buffered by the broker rather than immediately routed to queues.
3. **Commit or rollback** – Call `txCommit()` to atomically deliver all buffered messages, or `txRollback()` to discard them entirely.

The `AMQPChannel` class tracks transaction state through the `isTxMode: Boolean` property, allowing you to verify whether a channel is currently operating within a transaction boundary.

## Transaction API Layers in Kourier

Kourier exposes transaction capabilities through three architectural layers, each defined in specific source files within the `amqp-client` module.

### Low-Level Channel API

The core transaction methods reside in [`AMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/AMQPChannel.kt) at [`amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/AMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/AMQPChannel.kt) (lines 95-114). These are `suspend` functions that map directly to AMQP protocol methods:

- `txSelect()` – Enters transaction mode
- `txCommit()` – Commits the current transaction
- `txRollback()` – Aborts the current transaction

These functions operate on the raw channel and provide maximum control for atomic operations.

### DSL Extension Functions

For idiomatic Kotlin usage, [`Extensions.kt`](https://github.com/guimauvedigital/kourier/blob/main/Extensions.kt) at [`amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/Extensions.kt`](https://github.com/guimauvedigital/kourier/blob/main/amqp-client/src/commonMain/kotlin/dev/kourier/amqp/channel/Extensions.kt) (lines 326-382) provides fluent DSL wrappers:

- `txSelect { … }` – Wraps the transaction selection with a builder block
- `txCommit { … }` – Wraps the commit operation
- `txRollback { … }` – Wraps the rollback operation

These extensions return state objects (`SelectedTransactionMode`, `CommittedTransaction`, `RolledbackTransaction`) that provide type-safe representations of the transaction lifecycle.

### State Builders

The DSL layer utilizes builder classes defined in the `states` package:

- [`SelectedTransactionModeBuilder.kt`](https://github.com/guimauvedigital/kourier/blob/main/SelectedTransactionModeBuilder.kt) – Constructs `SelectedTransactionMode` instances
- [`CommittedTransactionBuilder.kt`](https://github.com/guimauvedigital/kourier/blob/main/CommittedTransactionBuilder.kt) – Constructs `CommittedTransaction` instances  
- [`RolledbackTransactionBuilder.kt`](https://github.com/guimauvedigital/kourier/blob/main/RolledbackTransactionBuilder.kt) – Constructs `RolledbackTransaction` instances

While these builders currently contain no configurable fields, they serve as extension points for future transaction options and maintain API consistency with other Kourier builders.

## Implementing Atomic Operations: Code Examples

The following examples demonstrate how to handle transactions in Kourier for atomic operations across different scenarios.

### Basic Transaction Pattern

This example shows the minimal implementation using the low-level API:

```kotlin
import dev.kourier.amqp.channel.AMQPChannel

suspend fun atomicPublish(channel: AMQPChannel) {
    // Enter transaction mode
    channel.txSelect()
    
    // Publish messages (buffered by broker)
    channel.basicPublish(
        body = "payload".toByteArray(),
        exchange = "my-exchange",
        routingKey = "routing.key"
    )
    
    // Commit atomically
    channel.txCommit()
}

```

### Exception-Driven Rollback

For production scenarios, wrap operations in try-catch blocks to ensure rollback on failure:

```kotlin
suspend fun safeBatchPublish(channel: AMQPChannel) {
    channel.txSelect()
    
    try {
        // Multiple publishes that must succeed together
        channel.basicPublish("first".toByteArray(), "ex", "rk")
        channel.basicPublish("second".toByteArray(), "ex", "rk")
        
        channel.txCommit()
    } catch (ex: Exception) {
        // Discard all buffered publishes
        channel.txRollback()
        throw ex
    }
}

```

### DSL Style Implementation

Use the extension functions for a more idiomatic Kotlin approach:

```kotlin
suspend fun dslStylePublish(channel: AMQPChannel) {
    channel.txSelect {
        // Builder block - currently no options available
    }
    
    try {
        channel.basicPublish("msg".toByteArray(), "ex", "rk")
        
        channel.txCommit {
            // Empty builder for future extensions
        }
    } catch (e: Exception) {
        channel.txRollback {
            // Empty builder
        }
        throw e
    }
}

```

## Error Handling and Safety Considerations

When you handle transactions in Kourier for atomic operations, implement these safety patterns:

**Verify transaction state** using the `isTxMode` property before committing:

```kotlin
if (channel.isTxMode) {
    channel.txCommit()
}

```

**Idempotent rollback** – Calling `txRollback()` when not in transaction mode is safe and treated as a no-op by the broker, though checking `isTxMode` prevents unnecessary network calls.

**Channel validity** – Transactions are bound to the channel lifecycle. If the channel closes due to an error, the transaction is automatically rolled back by the broker, and the `isTxMode` flag on the client side becomes stale. Always handle channel exceptions and recreate channels if necessary.

## Performance Considerations

Transactions in AMQP introduce latency because each `txSelect`, `txCommit`, or `txRollback` requires a synchronous round-trip to the broker. Additionally, buffered messages consume broker memory until committed.

To optimize performance when you handle transactions in Kourier for atomic operations:

- **Batch related messages** – Group multiple publishes into a single transaction to amortize the commit cost.
- **Avoid high-frequency transactions** – For high-throughput scenarios, consider publisher confirms instead of transactions, as confirms offer better performance with less broker overhead.
- **Keep transactions short** – Minimize the time between `txSelect` and `txCommit` to reduce memory pressure on the broker and exposure to network failures.

## Summary

- Kourier implements AMQP 0-9-1 transactions through `txSelect()`, `txCommit()`, and `txRollback()` in [`AMQPChannel.kt`](https://github.com/guimauvedigital/kourier/blob/main/AMQPChannel.kt).
- The **transaction mode** buffers publishes until commit, providing atomic delivery guarantees.
- **Two API layers** exist: low-level suspend functions and fluent DSL extensions in [`Extensions.kt`](https://github.com/guimauvedigital/kourier/blob/main/Extensions.kt).
- **State builders** (`SelectedTransactionMode`, `CommittedTransaction`, `RolledbackTransaction`) provide type-safe transaction lifecycle objects.
- Use `isTxMode` to verify transaction state before committing or rolling back.
- Transactions add latency and memory overhead; use them for atomicity requirements, not high-throughput scenarios.

## Frequently Asked Questions

### What happens if txCommit fails in Kourier?

If `txCommit()` fails due to a network error, broker rejection, or channel closure, the transaction is automatically rolled back by the broker and a `KourierAMQPException` is thrown. All buffered messages are discarded, and the channel is no longer in transaction mode (`isTxMode` becomes `false`). You should catch this exception and implement retry logic or dead-letter handling as appropriate for your application.

### Can I use transactions with consumer acknowledgments in Kourier?

Transactions in Kourier apply specifically to **publishing** operations via `basicPublish`. While AMQP technically supports transactional acknowledgments for consumers, Kourier's current implementation focuses on atomic publishing. For consumer-side atomicity, consider using manual acknowledgments (`autoAck = false`) with transaction-aware business logic, or implement idempotent consumers that can handle duplicate messages safely.

### How do I check if a Kourier channel is in transaction mode?

The `AMQPChannel` interface exposes the `isTxMode: Boolean` property, which returns `true` after a successful `txSelect()` call and `false` after `txCommit()`, `txRollback()`, or channel closure. Use this property to guard against invalid state transitions:

```kotlin
if (channel.isTxMode) {
    channel.txCommit()
}

```

### Are Kourier transactions supported across multiple channels?

No, AMQP transactions are **channel-scoped**, not connection-scoped. Each `AMQPChannel` maintains its own transaction state independent of other channels on the same connection. To achieve atomicity across multiple channels, you must implement distributed transaction patterns in your application logic (such as the Saga pattern) or use a single channel for all related publishes within the transaction boundary.