# How to Handle Connection Errors and Exceptions Gracefully in Kourier

> Master Kourier connection errors. Leverage RobustAMQPConnection for auto-reconnect, catch AMQPExceptions, and manage connection lifecycle effectively.

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

---

**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`](https://github.com/guimauvedigital/kourier/blob/main/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`](https://github.com/guimauvedigital/kourier/blob/main/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`](https://github.com/guimauvedigital/kourier/blob/main/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`](https://github.com/guimauvedigital/kourier/blob/main/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:

```kotlin
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:

```kotlin
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:

```kotlin
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.