# Error Handling Patterns (/learn/guides/error-handling)



<Callout type="info">
  Retries and dead-lettering build on KubeMQ's delivery semantics. Review [Delivery Guarantees](/learn/concepts/delivery-guarantees) in Fundamentals to understand at-least-once delivery and acknowledgement before designing your error-handling strategy.
</Callout>

## Error Categories [#error-categories]

KubeMQ errors fall into four categories. Handling differs by category, not by messaging pattern.

| Category          | Cause                                               | Example                                        | Recovery                        |
| ----------------- | --------------------------------------------------- | ---------------------------------------------- | ------------------------------- |
| **Validation**    | Invalid input before the message reaches the server | Empty channel name, body + metadata both empty | Fix the input — do not retry    |
| **Connection**    | Network or server unavailable                       | DNS failure, server restart, TLS mismatch      | Reconnect with backoff          |
| **Timeout**       | Operation exceeded its deadline                     | RPC request timeout, queue poll timeout        | Retry or increase timeout       |
| **Authorization** | Missing or invalid credentials                      | Bad auth token, expired certificate            | Refresh credentials, then retry |

## Connection Error Handling [#connection-error-handling]

Detect connection failures and attempt reconnection. All SDKs expose connection state or error callbacks.

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="connection_error.go"
    package main

    import (
        "context"
        "log"
        "time"

        "github.com/kubemq-io/kubemq-go/v2"
    )

    func connectWithRetry(ctx context.Context) *kubemq.Client {
        var client *kubemq.Client
        var err error
        backoff := time.Second

        for {
            client, err = kubemq.NewClient(ctx,
                kubemq.WithAddress("localhost", 50000),
                kubemq.WithClientId("resilient-client"),
            )
            if err == nil {
                if _, pingErr := client.Ping(ctx); pingErr == nil {
                    log.Println("Connected to KubeMQ")
                    return client
                }
                client.Close()
            }

            log.Printf("Connection failed: %v — retrying in %v", err, backoff)
            time.Sleep(backoff)
            if backoff < 30*time.Second {
                backoff *= 2
            }
        }
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python title="connection_error.py"
    import time
    from kubemq.pubsub import Client as PubSubClient

    def connect_with_retry(address: str) -> PubSubClient:
        backoff = 1.0
        while True:
            try:
                client = PubSubClient(
                    address=address,
                    client_id="resilient-client",
                )
                client.ping()
                print("Connected to KubeMQ")
                return client
            except Exception as e:
                print(f"Connection failed: {e} — retrying in {backoff}s")
                time.sleep(backoff)
                backoff = min(backoff * 2, 30)
    ```
  </Tab>

  <Tab value="Node.js">
    ```javascript title="connection_error.js"
    const { KubeMQClient } = require("kubemq-js");

    async function connectWithRetry(address) {
      let backoff = 1000;
      while (true) {
        try {
          const client = new KubeMQClient({
            address,
            clientId: "resilient-client",
          });
          await client.ping();
          console.log("Connected to KubeMQ");
          return client;
        } catch (err) {
          console.error(`Connection failed: ${err.message} — retrying in ${backoff}ms`);
          await new Promise((r) => setTimeout(r, backoff));
          backoff = Math.min(backoff * 2, 30000);
        }
      }
    }
    ```
  </Tab>

  <Tab value="Java">
    ```java title="ConnectionError.java"
    PubSubClient connectWithRetry(String address) throws InterruptedException {
        long backoff = 1000;
        while (true) {
            try {
                PubSubClient client = PubSubClient.builder()
                    .address(address)
                    .clientId("resilient-client")
                    .build();
                client.ping();
                System.out.println("Connected to KubeMQ");
                return client;
            } catch (Exception e) {
                System.err.printf("Connection failed: %s — retrying in %dms%n",
                    e.getMessage(), backoff);
                Thread.sleep(backoff);
                backoff = Math.min(backoff * 2, 30000);
            }
        }
    }
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="ConnectionError.cs"
    async Task<KubeMQClient> ConnectWithRetryAsync(string address)
    {
        var backoff = TimeSpan.FromSeconds(1);
        while (true)
        {
            try
            {
                var client = new KubeMQClient(new KubeMQClientOptions
                {
                    Address = address,
                    ClientId = "resilient-client",
                });
                await client.ConnectAsync();
                await client.PingAsync();
                Console.WriteLine("Connected to KubeMQ");
                return client;
            }
            catch (Exception ex)
            {
                Console.WriteLine($"Connection failed: {ex.Message} — retrying in {backoff}");
                await Task.Delay(backoff);
                if (backoff < TimeSpan.FromSeconds(30))
                    backoff *= 2;
            }
        }
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="ConnectionError.kt"
    import io.kubemq.sdk.pubsub.PubSubClient
    import kotlinx.coroutines.delay

    suspend fun connectWithRetry(address: String): PubSubClient {
        var backoff = 1000L
        while (true) {
            try {
                val client = PubSubClient(address)
                client.ping()
                println("Connected to KubeMQ")
                return client
            } catch (e: Exception) {
                println("Connection failed: ${e.message} — retrying in ${backoff}ms")
                delay(backoff)
                backoff = minOf(backoff * 2, 30000)
            }
        }
    }
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="connection_error.cpp"
    #include <kubemq/client.h>
    #include <iostream>
    #include <thread>
    #include <chrono>

    kubemq::PubSubClient connectWithRetry(const std::string& address) {
        int backoff = 1000;
        while (true) {
            try {
                kubemq::PubSubClient client(address);
                client.ping();
                std::cout << "Connected to KubeMQ" << std::endl;
                return client;
            } catch (const std::exception& e) {
                std::cerr << "Connection failed: " << e.what()
                          << " — retrying in " << backoff << "ms" << std::endl;
                std::this_thread::sleep_for(std::chrono::milliseconds(backoff));
                backoff = std::min(backoff * 2, 30000);
            }
        }
    }
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="connection_error.rs"
    use kubemq::prelude::*;
    use kubemq::RetryPolicy;
    use std::time::Duration;

    async fn connect_with_retry(host: &str, port: u16) -> KubemqClient {
        let mut backoff = Duration::from_secs(1);
        loop {
            // The builder applies its own retry policy on transient failures;
            // check_connection verifies the connection before returning.
            let result = KubemqClient::builder()
                .host(host)
                .port(port)
                .client_id("resilient-client")
                .check_connection(true)
                .retry_policy(RetryPolicy {
                    max_retries: 3,
                    initial_backoff: Duration::from_millis(100),
                    max_backoff: Duration::from_secs(10),
                    multiplier: 2.0,
                    ..Default::default()
                })
                .build()
                .await;

            match result {
                Ok(client) => {
                    println!("Connected to KubeMQ");
                    return client;
                }
                Err(e) => {
                    println!("Connection failed: {e} — retrying in {backoff:?}");
                    tokio::time::sleep(backoff).await;
                    backoff = std::cmp::min(backoff * 2, Duration::from_secs(30));
                }
            }
        }
    }
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="connection_error.rb"
    require 'kubemq'

    def connect_with_retry(address)
      backoff = 1.0
      loop do
        begin
          client = KubeMQ::PubSubClient.new(
            address: address,
            client_id: 'resilient-client'
          )
          client.ping
          puts 'Connected to KubeMQ'
          return client
        rescue KubeMQ::Error, StandardError => e
          puts "Connection failed: #{e.message} — retrying in #{backoff}s"
          sleep backoff
          backoff = [backoff * 2, 30].min
        end
      end
    end
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="connection_error.exs"
    # KubeMQ.Client.start_link/1 returns {:ok, client} or {:error, reason}.
    # Retry with exponential backoff until the connection succeeds.
    defmodule Resilient do
      def connect_with_retry(address, backoff \\ 1_000) do
        case KubeMQ.Client.start_link(address: address, client_id: "resilient-client") do
          {:ok, client} ->
            IO.puts("Connected to KubeMQ")
            client

          {:error, reason} ->
            IO.puts("Connection failed: #{inspect(reason)} — retrying in #{backoff}ms")
            Process.sleep(backoff)
            connect_with_retry(address, min(backoff * 2, 30_000))
        end
      end
    end
    ```
  </Tab>
</Tabs>

## Retry with Backoff [#retry-with-backoff]

Wrap send operations with exponential backoff and a maximum retry count to handle transient failures without overwhelming the server.

<Mermaid
  chart="stateDiagram-v2
    [*] --> Received
    Received --> Processing: deliver
    Processing --> Acked: success
    Processing --> Failed: error
    Failed --> Backoff: retries remain
    Backoff --> Requeued: wait elapsed
    Requeued --> Processing: redeliver
    Failed --> DeadLetter: retries exhausted
    Acked --> [*]
    DeadLetter --> [*]"
/>

*A message moves from received to processing; on failure it backs off and is requeued for redelivery, and once retries are exhausted it lands in the dead-letter queue.*

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="retry_backoff.go"
    func sendWithRetry(ctx context.Context, client *kubemq.Client, event *kubemq.Event, maxRetries int) error {
        backoff := 100 * time.Millisecond
        for attempt := 0; attempt <= maxRetries; attempt++ {
            err := client.SendEvent(ctx, event)
            if err == nil {
                return nil
            }
            if attempt == maxRetries {
                return fmt.Errorf("failed after %d retries: %w", maxRetries, err)
            }
            log.Printf("Attempt %d failed: %v — retrying in %v", attempt+1, err, backoff)
            time.Sleep(backoff)
            backoff *= 2
        }
        return nil
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python title="retry_backoff.py"
    import time

    def send_with_retry(client, event, max_retries=3):
        backoff = 0.1
        for attempt in range(max_retries + 1):
            try:
                client.send_event(event)
                return
            except Exception as e:
                if attempt == max_retries:
                    raise RuntimeError(f"Failed after {max_retries} retries") from e
                print(f"Attempt {attempt + 1} failed: {e} — retrying in {backoff}s")
                time.sleep(backoff)
                backoff *= 2
    ```
  </Tab>

  <Tab value="Node.js">
    ```javascript title="retry_backoff.js"
    async function sendWithRetry(client, event, maxRetries = 3) {
      let backoff = 100;
      for (let attempt = 0; attempt <= maxRetries; attempt++) {
        try {
          await client.sendEvent(event);
          return;
        } catch (err) {
          if (attempt === maxRetries) {
            throw new Error(`Failed after ${maxRetries} retries: ${err.message}`);
          }
          console.warn(`Attempt ${attempt + 1} failed: ${err.message} — retrying in ${backoff}ms`);
          await new Promise((r) => setTimeout(r, backoff));
          backoff *= 2;
        }
      }
    }
    ```
  </Tab>

  <Tab value="Java">
    ```java title="RetryBackoff.java"
    void sendWithRetry(PubSubClient client, EventMessage event, int maxRetries)
            throws Exception {
        long backoff = 100;
        for (int attempt = 0; attempt <= maxRetries; attempt++) {
            try {
                client.sendEventsMessage(event);
                return;
            } catch (Exception e) {
                if (attempt == maxRetries) {
                    throw new RuntimeException("Failed after " + maxRetries + " retries", e);
                }
                System.err.printf("Attempt %d failed: %s — retrying in %dms%n",
                    attempt + 1, e.getMessage(), backoff);
                Thread.sleep(backoff);
                backoff *= 2;
            }
        }
    }
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="RetryBackoff.cs"
    async Task SendWithRetryAsync(KubeMQClient client, EventMessage message, int maxRetries = 3)
    {
        var backoff = TimeSpan.FromMilliseconds(100);
        for (int attempt = 0; attempt <= maxRetries; attempt++)
        {
            try
            {
                await client.SendEventAsync(message);
                return;
            }
            catch (Exception ex)
            {
                if (attempt == maxRetries)
                    throw new InvalidOperationException($"Failed after {maxRetries} retries", ex);
                Console.WriteLine($"Attempt {attempt + 1} failed: {ex.Message} — retrying in {backoff}");
                await Task.Delay(backoff);
                backoff *= 2;
            }
        }
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="RetryBackoff.kt"
    suspend fun sendWithRetry(client: PubSubClient, event: EventMessage, maxRetries: Int = 3) {
        var backoff = 100L
        for (attempt in 0..maxRetries) {
            try {
                client.sendEvent(event)
                return
            } catch (e: Exception) {
                if (attempt == maxRetries) {
                    throw RuntimeException("Failed after $maxRetries retries", e)
                }
                println("Attempt ${attempt + 1} failed: ${e.message} — retrying in ${backoff}ms")
                delay(backoff)
                backoff *= 2
            }
        }
    }
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="retry_backoff.cpp"
    void sendWithRetry(kubemq::PubSubClient& client, kubemq::EventMessage& event, int maxRetries = 3) {
        int backoff = 100;
        for (int attempt = 0; attempt <= maxRetries; ++attempt) {
            try {
                client.sendEvent(event);
                return;
            } catch (const std::exception& e) {
                if (attempt == maxRetries) {
                    throw std::runtime_error(
                        "Failed after " + std::to_string(maxRetries) + " retries: " + e.what());
                }
                std::cerr << "Attempt " << attempt + 1 << " failed: " << e.what()
                          << " — retrying in " << backoff << "ms" << std::endl;
                std::this_thread::sleep_for(std::chrono::milliseconds(backoff));
                backoff *= 2;
            }
        }
    }
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="retry_backoff.rs"
    use kubemq::prelude::*;
    use kubemq::Event;
    use std::time::Duration;

    async fn send_with_retry(
        client: &KubemqClient,
        event: Event,
        max_retries: u32,
    ) -> kubemq::Result<()> {
        let mut backoff = Duration::from_millis(100);
        for attempt in 0..=max_retries {
            match client.send_event(event.clone()).await {
                Ok(()) => return Ok(()),
                Err(e) => {
                    if attempt == max_retries {
                        return Err(e);
                    }
                    println!("Attempt {} failed: {e} — retrying in {backoff:?}", attempt + 1);
                    tokio::time::sleep(backoff).await;
                    backoff *= 2;
                }
            }
        }
        Ok(())
    }
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="retry_backoff.rb"
    def send_with_retry(client, message, max_retries = 3)
      backoff = 0.1
      (0..max_retries).each do |attempt|
        begin
          client.send_event(message)
          return
        rescue KubeMQ::Error, StandardError => e
          raise "Failed after #{max_retries} retries: #{e.message}" if attempt == max_retries

          puts "Attempt #{attempt + 1} failed: #{e.message} — retrying in #{backoff}s"
          sleep backoff
          backoff *= 2
        end
      end
    end
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="retry_backoff.exs"
    # send_event/2 returns :ok or {:error, err}. Retry transient failures
    # with exponential backoff, raising once retries are exhausted.
    defmodule Retry do
      def send_with_retry(client, event, max_retries \\ 3, backoff \\ 100, attempt \\ 0) do
        case KubeMQ.Client.send_event(client, event) do
          :ok ->
            :ok

          {:error, err} when attempt >= max_retries ->
            raise "Failed after #{max_retries} retries: #{err.message}"

          {:error, err} ->
            IO.puts("Attempt #{attempt + 1} failed: #{err.message} — retrying in #{backoff}ms")
            Process.sleep(backoff)
            send_with_retry(client, event, max_retries, backoff * 2, attempt + 1)
        end
      end
    end
    ```
  </Tab>
</Tabs>

## Graceful Shutdown [#graceful-shutdown]

Close client connections cleanly to flush pending messages and release server resources.

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="graceful_shutdown.go"
    package main

    import (
        "context"
        "log"
        "os"
        "os/signal"
        "syscall"

        "github.com/kubemq-io/kubemq-go/v2"
    )

    func main() {
        ctx, cancel := context.WithCancel(context.Background())
        client, err := kubemq.NewClient(ctx,
            kubemq.WithAddress("localhost", 50000),
        )
        if err != nil {
            log.Fatal(err)
        }

        sig := make(chan os.Signal, 1)
        signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)

        go func() {
            <-sig
            log.Println("Shutting down...")
            cancel()
            client.Close()
            os.Exit(0)
        }()

        log.Println("Running — press Ctrl+C to stop")
        <-ctx.Done()
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python title="graceful_shutdown.py"
    import signal
    import sys
    from kubemq.pubsub import Client as PubSubClient

    client = PubSubClient(address="localhost:50000")

    def shutdown(signum, frame):
        print("Shutting down...")
        client.close()
        sys.exit(0)

    signal.signal(signal.SIGINT, shutdown)
    signal.signal(signal.SIGTERM, shutdown)

    print("Running — press Ctrl+C to stop")
    signal.pause()
    ```
  </Tab>

  <Tab value="Node.js">
    ```javascript title="graceful_shutdown.js"
    const { KubeMQClient } = require("kubemq-js");

    const client = new KubeMQClient({ address: "localhost:50000" });

    async function shutdown() {
      console.log("Shutting down...");
      await client.close();
      process.exit(0);
    }

    process.on("SIGINT", shutdown);
    process.on("SIGTERM", shutdown);

    console.log("Running — press Ctrl+C to stop");
    ```
  </Tab>

  <Tab value="Java">
    ```java title="GracefulShutdown.java"
    PubSubClient client = PubSubClient.builder()
        .address("localhost:50000")
        .clientId("shutdown-demo")
        .build();

    Runtime.getRuntime().addShutdownHook(new Thread(() -> {
        System.out.println("Shutting down...");
        client.close();
    }));

    System.out.println("Running — press Ctrl+C to stop");
    Thread.currentThread().join();
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="GracefulShutdown.cs"
    await using var client = new KubeMQClient(new KubeMQClientOptions
    {
        Address = "localhost:50000",
    });
    await client.ConnectAsync();

    var cts = new CancellationTokenSource();
    Console.CancelKeyPress += (_, e) =>
    {
        e.Cancel = true;
        Console.WriteLine("Shutting down...");
        cts.Cancel();
    };

    Console.WriteLine("Running — press Ctrl+C to stop");
    try { await Task.Delay(Timeout.Infinite, cts.Token); }
    catch (OperationCanceledException) { }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="GracefulShutdown.kt"
    val client = PubSubClient("localhost:50000")

    Runtime.getRuntime().addShutdownHook(Thread {
        println("Shutting down...")
        client.close()
    })

    println("Running — press Ctrl+C to stop")
    Thread.currentThread().join()
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="graceful_shutdown.cpp"
    #include <kubemq/client.h>
    #include <csignal>
    #include <iostream>
    #include <atomic>

    std::atomic<bool> running{true};
    kubemq::PubSubClient* globalClient = nullptr;

    void signalHandler(int) {
        std::cout << "Shutting down..." << std::endl;
        running = false;
        if (globalClient) globalClient->close();
    }

    int main() {
        kubemq::PubSubClient client("localhost:50000");
        globalClient = &client;

        std::signal(SIGINT, signalHandler);
        std::signal(SIGTERM, signalHandler);

        std::cout << "Running — press Ctrl+C to stop" << std::endl;
        while (running) {
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
        return 0;
    }
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="graceful_shutdown.rs"
    use kubemq::prelude::*;

    #[tokio::main]
    async fn main() -> kubemq::Result<()> {
        let client = KubemqClient::builder()
            .host("localhost")
            .port(50000)
            .build()
            .await?;

        println!("Running — press Ctrl+C to stop");

        // Wait for a shutdown signal, then close the client cleanly.
        tokio::signal::ctrl_c().await.ok();
        println!("Shutting down...");

        // close() flushes pending work and cancels active subscriptions.
        client.close().await?;
        println!("Client closed — shutdown complete");

        Ok(())
    }
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="graceful_shutdown.rb"
    require 'kubemq'

    client = KubeMQ::PubSubClient.new(address: 'localhost:50000', client_id: 'shutdown-demo')

    shutdown = proc do
      puts 'Shutting down...'
      client.close
      exit 0
    end

    Signal.trap('INT', &shutdown)
    Signal.trap('TERM', &shutdown)

    puts 'Running — press Ctrl+C to stop'
    sleep
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="graceful_shutdown.exs"
    {:ok, client} =
      KubeMQ.Client.start_link(address: "localhost:50000", client_id: "shutdown-demo")

    # Trap exits so the process can close the client before terminating.
    Process.flag(:trap_exit, true)
    IO.puts("Running — press Ctrl+C to stop")

    receive do
      {:EXIT, _from, _reason} ->
        IO.puts("Shutting down...")
        KubeMQ.Client.close(client)
        IO.puts("Client closed — shutdown complete")
    end
    ```
  </Tab>
</Tabs>

## Error Code Reference [#error-code-reference]

Common error codes returned by the KubeMQ server across all patterns.

| Code | Category      | Description                                                    |
| ---- | ------------- | -------------------------------------------------------------- |
| 100  | Validation    | Message body and metadata are both empty                       |
| 107  | Validation    | Channel name contains wildcard characters                      |
| 108  | Validation    | Channel name contains whitespace                               |
| 119  | Validation    | Channel name ends with a dot                                   |
| 120  | Validation    | Channel name is empty                                          |
| 200  | Connection    | Server unavailable or connection refused                       |
| 201  | Connection    | Connection timeout exceeded                                    |
| 300  | Timeout       | Request timeout (RPC commands/queries)                         |
| 301  | Timeout       | Queue poll wait timeout (not an error — no messages available) |
| 400  | Authorization | Authentication token missing or invalid                        |
| 401  | Authorization | Client not authorized for the requested channel                |

<Callout type="info">
  Code 301 (queue poll timeout) is expected behavior when no messages are available. Do not treat it as a failure in your error handling logic.
</Callout>