Overview

A graceful shutdown stops KubeMQ clients without dropping in-flight messages or leaking server-side subscription state. Killing a process outright, or closing a client mid-callback, can truncate a handler or leave the server thinking a consumer is still there. In a container platform that sends SIGTERM before force-killing a pod, an orderly shutdown sequence turns a rolling deploy into a clean handoff instead of a burst of errors.

The pattern is a fixed sequence: cancel subscriptions first so no new messages arrive, give pending operations a brief window to drain, then close each client. subJob.cancel() stops the subscription's Flow collection, and structured concurrency ensures resources tied to that coroutine's scope are released. When an application holds several clients — a pubSub, queues, and cq client — they're closed in reverse construction order, so one client that might still reference another isn't torn down first.

Gotchas: a fixed delay() after cancelling is a placeholder, not a guarantee — track in-flight work with a Job you explicitly join() instead. Closing clients out of order can leave one trying to use a transport another already shut down.

Prerequisites

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

Code

GracefulShutdownExample.kt

package io.kubemq.sdk.examples.errorhandling

import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.pubsub.eventMessage
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-error-handling-graceful-shutdown"
private const val CHANNEL = "kotlin-error-handling.graceful-shutdown"

fun main() = runBlocking {
    println("=== Graceful Shutdown ===\n")

    // Create multiple clients
    val pubSubClient = KubeMQClient.pubSub {
        address = ADDRESS
        clientId = "$CLIENT_ID-pubsub"
    }
    val queuesClient = KubeMQClient.queues {
        address = ADDRESS
        clientId = "$CLIENT_ID-queues"
    }
    val cqClient = KubeMQClient.cq {
        address = ADDRESS
        clientId = "$CLIENT_ID-cq"
    }

    println("Created 3 clients.")

    // Create channel and subscribe
    pubSubClient.createEventsChannel(CHANNEL)
    val subJob: Job = launch {
        pubSubClient.subscribeToEvents {
            channel = CHANNEL
        }.collect { /* process events */ }
    }
    println("Subscription active.\n")

    // Send messages before shutdown
    repeat(3) { i ->
        pubSubClient.publishEvent(eventMessage {
            channel = CHANNEL
            body = "Message ${i + 1}".toByteArray()
        })
    }
    delay(300)

    // Graceful shutdown sequence
    println("--- Initiating Graceful Shutdown ---\n")

    // Step 1: Cancel subscriptions
    subJob.cancel()
    println("1. Subscriptions cancelled.")

    // Step 2: Wait for pending operations
    delay(200)
    println("2. Pending operations completed.")

    // Step 3: Clean up channels
    try { pubSubClient.deleteEventsChannel(CHANNEL) } catch (_: Exception) {}
    println("3. Channels cleaned up.")

    // Step 4: Close clients in reverse order
    cqClient.close()
    queuesClient.close()
    pubSubClient.close()
    println("4. All clients closed.\n")

    println("Graceful shutdown complete.")
}

How It Works

Step 1: Cancel all subscription coroutine jobs to stop receiving messages.
Step 2: Allow a brief delay for pending send/receive operations to complete.
Step 3: Clean up channels if needed (optional, depending on use case).
Step 4: Close clients in reverse creation order to avoid dependency issues.
Kotlin's structured concurrency ensures that cancelling a job properly cleans up resources.

Graceful Shutdown

Overview

Prerequisites

Code

How It Works

Related

On this page