How to Handle Connection Errors and Exceptions Gracefully in Kourier

Use RobustAMQPConnection for automatic reconnection, catch typed AMQPException hierarchies for granular error handling, and observe the connectionClosed deferred for lifecycle management.

Kourier is a Kotlin multiplatform AMQP client that separates connection lifecycles from error propagation to help you handle connection errors and exceptions gracefully in Kourier. The library provides distinct layers for low-level socket management, typed exception hierarchies, and robust automatic reconnection with state recovery.

Understanding Kourier's Connection Error Architecture

Kourier's error handling strategy divides responsibilities between the core connection implementation and exception types. This separation allows you to choose between manual error handling or automatic recovery depending on your reliability requirements.

DefaultAMQPConnection and the Connection Lifecycle

The DefaultAMQPConnection class in amqp-client/src/commonMain/kotlin/dev/kourier/amqp/connection/DefaultAMQPConnection.kt manages the raw TCP socket and AMQP frame decoding. It tracks connection states (OPEN, CLOSED, SHUTTING_DOWN) and exposes a connectionClosed property—a CompletableDeferred that completes with an AMQPException.ConnectionClosed when the connection terminates.

Socket-level failures (such as IOException or EOFException) are caught in the startListening method and converted into broker-style close frames via closeFromChannelException. This ensures that even unexpected network errors result in a clean state transition and notification through the connectionClosed deferred.

Broker-initiated shutdowns are handled by closeFromBroker, which transitions the state to SHUTTING_DOWN, cancels all active channels, and emits the close reason. Graceful client-side closure uses the close(reason, code) method to send a Connection.Close frame and await the broker's CloseOk response.

The AMQPException Hierarchy

Kourier defines a comprehensive exception hierarchy in amqp-client/src/commonMain/kotlin/dev/kourier/amqp/AMQPException.kt. This typing allows you to distinguish between recoverable and fatal errors:

• AMQPException.ConnectionClosed: Thrown when the underlying TCP connection terminates, either from network failures or broker shutdown.
• AMQPException.ChannelClosed: Indicates a specific channel was closed, often due to protocol errors or resource constraints.
• AMQPException.InvalidUrl: Configuration error when parsing connection strings.

Catching these specific types allows you to implement targeted recovery logic—retrying connections for network failures while escalating configuration errors immediately.

Automatic Reconnection with RobustAMQPConnection

For production services requiring high availability, Kourier provides the amqp-client-robust module. The RobustAMQPConnection class wraps the core connection with automatic reconnection and state restoration logic.

The Connection Factory Loop

The RobustAMQPConnection.connectionFactory() method (located in amqp-client-robust/src/commonMain/kotlin/dev/kourier/amqp/robust/RobustAMQPConnection.kt) implements a long-running coroutine that manages the connection lifecycle:

1. Connection Attempt: Calls super.connect() to establish the TCP socket.
2. Channel Restoration: Invokes channel.restore() on every RobustAMQPChannel to rebuild AMQP state.
3. Await Closure: Suspends until closedResponses.first() indicates broker-initiated shutdown.
4. Exception Handling: Catches any exception, logs the failure, and continues the loop to trigger reconnection.
5. Termination: Exits only when connectionClosed completes, indicating permanent shutdown.

This loop ensures that transient network failures result in automatic reconnection without application-level intervention.

Channel Restoration and State Recovery

The RobustAMQPChannel class (in amqp-client-robust/src/commonMain/kotlin/dev/kourier/amqp/robust/RobustAMQPChannel.kt) extends standard channels with state-preserving logic. When the broker closes a channel, the cancelAll method triggers restoration only if the underlying connection remains OPEN.

The restoration process re-issues all previously declared queues, exchanges, bindings, QoS settings, and consumer registrations. To prevent acknowledgment failures after reconnection, the implementation filters out stale delivery-tags from the internal tracking maps before restoring consumers. This avoids PRECONDITION_FAILED errors that would otherwise occur when acknowledging messages from the previous connection session.

Practical Implementation Examples

Basic Error Handling with RobustAMQPConnection

Use RobustAMQPConnection for production services that require automatic recovery from network partitions:

suspend fun runConsumer() {
    val config = AMQPConfig(
        server = AMQPConfig.Server(
            host = "broker.example.com",
            port = 5672,
            user = "guest",
            password = "guest"
        )
    )
    
    val connection = RobustAMQPConnection.create(CoroutineScope(Dispatchers.IO), config)

    try {
        val channel = connection.createChannel()
        channel.basicConsume(
            queue = "my-queue",
            consumerTag = "",
            noAck = false,
            exclusive = false,
            arguments = Table.empty(),
            onDelivery = { delivery ->
                println("Message ${delivery.message.bodyString()}")
                channel.basicAck(delivery.message.deliveryTag, false)
            },
            onCanceled = { /* handle consumer cancel */ }
        )
    } catch (e: AMQPException.ConnectionClosed) {
        println("Connection lost: ${e.replyText}")
    } catch (e: AMQPException.ChannelClosed) {
        println("Channel closed: ${e.replyText}")
    }

    connection.connectionClosed.invokeOnCompletion { cause ->
        if (cause != null) println("Connection terminated: ${cause.message}")
    }
}

Manual Connection Management

For scenarios requiring explicit control over reconnection logic, use DefaultAMQPConnection directly:

suspend fun runWithPlainConnection() {
    val conn = DefaultAMQPConnection.create(CoroutineScope(Dispatchers.IO), config)
    try {
        // Normal usage
        val channel = conn.createChannel()
        // ... message processing ...
    } finally {
        conn.close("normal shutdown", 200u)
    }

    runCatching { conn.connectionClosed.await() }
        .onFailure { ex -> println("Unexpected close: ${ex?.message}") }
}

Custom Reconnection Strategies

Implement exponential backoff or circuit breakers by wrapping the connection factory loop:

suspend fun customReconnect() {
    var connection: AMQPConnection? = null
    var attempt = 0

    while (connection == null || connection.connectionClosed.isCompleted.not()) {
        try {
            connection = DefaultAMQPConnection.create(CoroutineScope(Dispatchers.IO), config)
            connection.connectionClosed.await()
        } catch (e: Exception) {
            attempt++
            val delayMs = (2.0.pow(attempt.toDouble()) * 1000L).toLong()
            println("Reconnect attempt $attempt failed: ${e.message}. Back‑off $delayMs ms")
            delay(delayMs)
        }
    }
}

Summary

• DefaultAMQPConnection provides low-level socket management and exposes connectionClosed as a CompletableDeferred for lifecycle observation.
• AMQPException hierarchy distinguishes between ConnectionClosed, ChannelClosed, and configuration errors, enabling targeted catch blocks.
• RobustAMQPConnection implements automatic reconnection via connectionFactory(), restoring all channels and consumers after network failures.
• RobustAMQPChannel preserves AMQP state (queues, bindings, QoS) across reconnections and filters stale delivery-tags to prevent acknowledgment errors.
• For production services, prefer RobustAMQPConnection; for custom retry logic, use DefaultAMQPConnection with manual connectionClosed observation.

Frequently Asked Questions

What happens when the broker closes a connection unexpectedly?

When the broker initiates a shutdown, DefaultAMQPConnection.closeFromBroker transitions the state to SHUTTING_DOWN, cancels all active channels, and completes the connectionClosed deferred with an AMQPException.ConnectionClosed. If using RobustAMQPConnection, the connectionFactory loop catches this completion and immediately attempts to reconnect, restoring all channels and their state.

How does RobustAMQPConnection handle channel failures?

RobustAMQPConnection wraps channels with RobustAMQPChannel, which intercepts Channel.Close frames via cancelAll. If the underlying connection remains OPEN, the channel automatically enters a restoration phase that re-declares queues, exchanges, bindings, QoS settings, and consumer registrations. It also purges stale delivery-tags from internal tracking maps to prevent PRECONDITION_FAILED errors when acknowledging messages after reconnection.

Can I implement custom backoff strategies with Kourier?

Yes. While RobustAMQPConnection provides immediate reconnection, you can implement custom strategies using DefaultAMQPConnection directly. Create a wrapper loop that catches connection failures, applies exponential backoff (e.g., delay(2.0.pow(attempt) * 1000)), and re-instantiates the connection. Observe the connectionClosed deferred to distinguish between retryable network errors and permanent broker shutdowns.

What's the difference between ConnectionClosed and ChannelClosed exceptions?

AMQPException.ConnectionClosed indicates the underlying TCP socket has terminated, affecting all channels on that connection. This is typically thrown from DefaultAMQPConnection when the broker sends a Connection.Close frame or when a socket error occurs. AMQPException.ChannelClosed indicates a specific AMQP channel (a virtual connection within the TCP socket) has closed, often due to protocol errors, resource constraints, or queue/exchange declaration conflicts. RobustAMQPConnection automatically recovers from both, but distinguishing them allows you to log severity appropriately—connection failures usually indicate network issues while channel failures often indicate application logic errors.

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