How to Implement Publisher Confirms in Kourier for Reliable Message Publishing

Call channel.confirmSelect() to enable RabbitMQ publisher confirm mode, then consume the cold Flow from publishConfirmResponses to receive Ack or Nack objects that indicate whether the broker successfully persisted each message.

Kourier (guimauvedigital/kourier) is a Kotlin multiplatform AMQP client that implements the RabbitMQ Publisher-Confirm pattern for reliable message delivery. When you implement publisher confirms in Kourier, the broker asynchronously notifies your application whether each message reached the queue, enabling exactly-once semantics or high-throughput batch processing. This mechanism is essential for building fault-tolerant systems where message loss is unacceptable.

Enabling Confirm Mode on the Channel

Before publishing messages with delivery guarantees, you must place the channel into confirm mode using the confirmSelect() method.

The confirmSelect() Method

In dev/kourier/amqp/channel/DefaultAMQPChannel.kt (lines 46-57), the confirmSelect() implementation sends the Confirm.Select RPC command to the broker and atomically flips the isConfirmMode boolean to true:

suspend fun confirmSelect() {
    // Sends Confirm.Select frame to broker
    rpc(AMQPMethod.Channel.Confirm.Select(nowait = false))
    isConfirmMode = true
}

Once enabled, every subsequent basicPublish call returns a Published response containing a monotonically increasing deliveryTag. This tag is generated atomically using a Mutex to ensure thread safety across concurrent coroutines (lines 74-80).

The publishConfirmResponses Flow

The AMQPChannel interface declares publishConfirmResponses as a cold Flow<AMQPResponse.Channel.Basic.PublishConfirm> (lines 36-41 in AMQPChannel.kt). The default implementation in DefaultAMQPChannel.kt (lines 49-51) filters the internal MutableSharedFlow<AMQPResponse> to emit only confirm frames:

override val publishConfirmResponses: Flow<AMQPResponse.Channel.Basic.PublishConfirm>
    get() = channelResponses.filterIsInstance()

Because this is a cold flow, you control when to start collecting confirms. The flow emits either Ack (broker persisted the message) or Nack (broker rejected or failed to process) objects, each carrying the deliveryTag and a multiple flag.

How Publisher Confirms Work Internally

When isConfirmMode is active, the publishing workflow follows these steps:

1. Tag Generation – basicPublish increments an internal deliveryTag counter protected by a Mutex and returns it in the Published response.
2. Broker Acknowledgment – The RabbitMQ broker later sends a Basic.Ack or Basic.Nack frame containing the same delivery tag.
3. Frame Decoding – Kourier's connection decoder transforms these frames into AMQPResponse.Channel.Basic.PublishConfirm.Ack or .Nack instances and pushes them into the shared channelResponses flow.
4. Client Consumption – Your collector receives these confirms via publishConfirmResponses, allowing you to correlate them with the original publish operation using the deliveryTag.

The multiple flag indicates that the broker is acknowledging all messages up to and including the specified delivery tag in a single frame, which is useful for optimizing high-throughput scenarios.

Implementation Strategies for Kourier Publisher Confirms

Kourier supports three primary patterns for handling confirms, each balancing latency against throughput.

Strategy 1: Individual Publish-and-Wait (Synchronous)

For strict exactly-once guarantees where you must confirm persistence before proceeding, publish a single message and block until the corresponding confirm arrives. This pattern is demonstrated in PublisherConfirmsTest.kt (lines 20-58):

channel.confirmSelect()

for (i in 1..messageCount) {
    channel.basicPublish(
        "Message $i".toByteArray(),
        exchange = "",
        routingKey = queueName,
        properties = Properties()
    )

    // Block until the broker acknowledges this single message
    when (val confirm = channel.publishConfirmResponses.first()) {
        is AMQPResponse.Channel.Basic.PublishConfirm.Ack  -> println("✅ $i acked")
        is AMQPResponse.Channel.Basic.PublishConfirm.Nack -> println("❌ $i nacked")
    }
}

When to use: Low-volume, critical financial transactions or command messages where durability is more important than throughput.

Strategy 2: Batch Publishing for High Throughput

To reduce network round-trips, publish a batch of messages and wait for the entire batch to be acknowledged. This approach is implemented in PublisherConfirmsTest.kt (lines 90-120):

channel.confirmSelect()
val batchSize = 10

while (published < totalMessages) {
    repeat(batchSize) {
        channel.basicPublish(
            "Batch ${published + it + 1}".toByteArray(),
            exchange = "",
            routingKey = queueName,
            properties = Properties()
        )
    }

    // Collect exactly `batchSize` confirm frames
    val confirms = channel.publishConfirmResponses
        .take(batchSize)
        .toList()

    val acks = confirms.count { it is AMQPResponse.Channel.Basic.PublishConfirm.Ack }
    val nacks = confirms.size - acks
    println("Batch $published: $acks acks, $nacks nacks")
    published += batchSize
}

When to use: High-volume logging, telemetry, or event streaming where you can tolerate small windows of unconfirmed messages in memory.

Strategy 3: Asynchronous Confirm Handling

For fire-and-forget publishing with background verification, launch a coroutine that collects confirms independently of your publishing loop. This pattern (lines 124-173 in PublisherConfirmsTest.kt) allows maximum publisher throughput:

channel.confirmSelect()
val messageCount = 20
val confirmed = mutableListOf<ULong>()
val nacked = mutableListOf<ULong>()

// Coroutine that consumes confirms as they arrive
val confirmJob = launch {
    channel.publishConfirmResponses
        .take(messageCount)
        .collect { confirm ->
            when (confirm) {
                is AMQPResponse.Channel.Basic.PublishConfirm.Ack -> {
                    confirmed.add(confirm.deliveryTag)
                    println("✅ ack ${confirm.deliveryTag}")
                }
                is AMQPResponse.Channel.Basic.PublishConfirm.Nack -> {
                    nacked.add(confirm.deliveryTag)
                    println("❌ nack ${confirm.deliveryTag}")
                }
            }
        }
}

// Fire-and-forget publishing
repeat(messageCount) { i ->
    channel.basicPublish(
        "Async $i".toByteArray(),
        exchange = "",
        routingKey = queueName,
        properties = Properties()
    )
}

confirmJob.join()
println("Done – ${confirmed.size} acks, ${nacked.size} nacks")

When to use: Real-time systems where you cannot block the publisher but need to log or retry failed messages asynchronously.

Handling Bulk Confirms with the Multiple Flag

RabbitMQ can acknowledge ranges of messages using the multiple flag to reduce network traffic. When multiple is true, the delivery tag represents the highest tag in a contiguous range of confirmed messages. The test suite (lines 184-206) demonstrates how to detect and handle these bulk confirms:

val confirms = channel.publishConfirmResponses.take(messageCount).toList()
val multi = confirms.filter { it.multiple }
if (multi.isNotEmpty()) {
    println("Received ${multi.size} bulk confirms (multiple = true)")
}
assertTrue(confirms.all { it is AMQPResponse.Channel.Basic.PublishConfirm.Ack })

When processing confirms with multiple = true, you should consider all delivery tags less than or equal to the specified tag as acknowledged.

Summary

• Enable confirms by calling confirmSelect() on the channel, which sets isConfirmMode = true in DefaultAMQPChannel.kt.
• Consume confirms via the cold Flow at publishConfirmResponses, which filters the internal channelResponses stream for Ack and Nack frames.
• Correlate messages using the deliveryTag returned by basicPublish, which is atomically incremented and protected by a Mutex.
• Choose your strategy based on reliability requirements: individual blocking confirms for safety, batch confirms for throughput, or asynchronous collection for maximum performance.
• Handle bulk confirms by checking the multiple flag, which indicates that all messages up to the delivery tag are acknowledged.

Frequently Asked Questions

How does Kourier generate delivery tags for publisher confirms?

According to the source code in DefaultAMQPChannel.kt (lines 74-80), Kourier maintains an internal deliveryTag counter as a ULong. Each call to basicPublish while isConfirmMode is true atomically increments this counter using a Mutex to prevent race conditions in concurrent coroutines. The incremented value is returned in the Published response, allowing you to match it against the deliveryTag in the subsequent Ack or Nack.

What is the difference between Ack and Nack in Kourier's publishConfirmResponses?

AMQPResponse.Channel.Basic.PublishConfirm.Ack indicates that the broker successfully received and persisted the message to disk or a queue, while AMQPResponse.Channel.Basic.PublishConfirm.Nack signals that the broker was unable to process the message due to resource constraints, queue limits, or internal errors. Your application should handle Nack by logging the failure, retrying the publish, or routing the message to a dead-letter queue based on the deliveryTag.

What happens if I publish without calling confirmSelect()?

If you call basicPublish without first enabling confirm mode via confirmSelect(), the channel operates in standard fire-and-forget mode. The basicPublish method will not return a deliveryTag, and publishConfirmResponses will remain empty because the broker does not send Basic.Ack or Basic.Nack frames for non-confirm channels. You lose the ability to verify message durability at the application level.

How do I handle the multiple flag when receiving confirms?

When PublishConfirm.multiple is true, the broker is acknowledging all messages up to and including the specified deliveryTag in a single frame. In Kourier, you process this by checking the flag on each confirm object and, if true, removing all pending tags less than or equal to that delivery tag from your internal tracking set. This optimization reduces the number of frames the broker must send during high-throughput scenarios.

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