How to Handle Transactions in Kourier for Atomic Operations

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 at 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 at 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 – Constructs SelectedTransactionMode instances
• CommittedTransactionBuilder.kt – Constructs CommittedTransaction instances
• 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:

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:

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:

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:

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.
• 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.
• 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:

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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}