Overview

A network partition, a server that hasn't started yet, or a typo in the address are all normal facts of life in distributed systems — and a client that blocks indefinitely or crashes with an unhandled exception turns a routine outage into a cascading failure. Fail-fast connection checking lets you detect an unreachable KubeMQ server the moment you ping() a freshly built client, so your service can log the failure, alert, or fall back instead of hanging.

KubeMQException is a sealed hierarchy — Connection, Validation, Authentication, Authorization, Timeout, Throttling, Transport, Server, ClientClosed, StreamBroken — so a when expression on the caught exception can exhaustively branch by failure type instead of string-matching a message. Every variant also exposes isRetryable, letting you decide programmatically whether to retry or fail immediately. Gotchas: the compiler only enforces exhaustiveness on the sealed type itself — wrapping it in a broader catch (e: Exception) silently reintroduces the cases you meant to handle explicitly; isRetryable reflects the failure category, not your retry budget, so retrying a dead server in a tight loop just multiplies the outage; a successful ping() doesn't guarantee the connection stays healthy, so mid-session drops still need reconnection handling.

Prerequisites

KubeMQ server running on localhost:50000
Kotlin SDK installed (implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))

Code

ConnectionErrorExample.kt

package io.kubemq.sdk.examples.errorhandling

import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.exception.KubeMQException
import io.kubemq.sdk.pubsub.eventMessage
import kotlinx.coroutines.runBlocking

private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-error-handling-connection-error"

fun main() = runBlocking {
    // 1. Bad address error
    println("=== Bad Address Error ===\n")
    try {
        val badClient = KubeMQClient.pubSub {
            address = "localhost:59999"
            clientId = CLIENT_ID
        }
        badClient.use {
            it.ping()
        }
    } catch (e: KubeMQException) {
        when (e) {
            is KubeMQException.Connection ->
                println("Connection error (expected): ${e.message}")
            else ->
                println("Error: ${e.message} (code=${e.code}, category=${e.category})")
        }
        println("  retryable=${e.isRetryable}")
    } catch (e: Exception) {
        println("Error (expected): ${e.javaClass.simpleName}: ${e.message}")
    }

    // 2. Successful connection
    println("\n=== Successful Connection ===\n")
    val client = KubeMQClient.pubSub {
        address = ADDRESS
        clientId = CLIENT_ID
    }

    client.use {
        val info = client.ping()
        println("Connected to ${info.host} v${info.version}")

        // 3. Typed exception handling
        println("\n=== Typed Exception Handling ===\n")
        try {
            client.publishEvent(eventMessage {
                channel = "" // invalid
                body = "test".toByteArray()
            })
        } catch (e: KubeMQException) {
            when (e) {
                is KubeMQException.Validation ->
                    println("Validation: ${e.message} (op=${e.operation})")
                is KubeMQException.Connection ->
                    println("Connection: ${e.message} (retryable=${e.isRetryable})")
                is KubeMQException.Authentication ->
                    println("Auth: ${e.message}")
                is KubeMQException.Authorization ->
                    println("Authz: ${e.message} (channel=${e.channel})")
                is KubeMQException.Timeout ->
                    println("Timeout: ${e.message} (duration=${e.duration})")
                is KubeMQException.Throttling ->
                    println("Throttled: ${e.message}")
                is KubeMQException.Transport ->
                    println("Transport: ${e.message}")
                is KubeMQException.Server ->
                    println("Server: ${e.message} (code=${e.statusCode})")
                is KubeMQException.ClientClosed ->
                    println("Client closed: ${e.message}")
                is KubeMQException.StreamBroken ->
                    println("Stream broken: ${e.message}")
            }
            println("  code=${e.code}, category=${e.category}, retryable=${e.isRetryable}")
        }

        // 4. Retryable check
        println("\n=== Retryable Check ===\n")
        try {
            client.publishEvent(eventMessage {
                channel = "kotlin-error-handling.connection-test"
                body = "test".toByteArray()
            })
            println("Published successfully")
        } catch (e: KubeMQException) {
            if (e.isRetryable) {
                println("Error is retryable (${e.code}), would retry...")
            } else {
                println("Error is NOT retryable (${e.code}), failing immediately")
            }
        }

        println("\nDone.")
    }
}

How It Works

The sealed KubeMQException hierarchy enables exhaustive when expressions for type-safe error handling.
Each exception type has specific properties (e.g., Timeout.duration, Authorization.channel, Validation.operation).
isRetryable indicates whether the operation can be safely retried.
Connection errors to invalid addresses are caught and identified as retryable.
Validation errors (like empty channel names) are caught as non-retryable.

Connection Error

Overview

Prerequisites

Code

How It Works

On this page

Connection Error

Overview

Prerequisites

Code

How It Works

Related

On this page