Graceful Shutdown
Shut down a KubeMQ client cleanly in Go, unsubscribing and releasing resources before exit for production-safe operation.
Overview
A graceful shutdown stops a KubeMQ client without dropping in-flight messages or leaking server-side subscription state. Killing a process outright, or closing the connection 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, handling that signal turns a rolling deploy into a clean handoff instead of a burst of errors.
The pattern has a fixed order: stop new work by unsubscribing, then close the client with a drain timeout so in-flight operations get a bounded window to finish before the gRPC connection is torn down. signal.NotifyContext wires SIGINT/SIGTERM into a cancellable context.Context, and kubemq.WithDrainTimeout sets how long Close() waits. kubemq.WithOnClosed gives you a hook to flush metrics once the connection is fully down.
Gotchas: unsubscribing after closing the client, instead of before, can race the connection teardown. A drain timeout that's too short cuts off the message you were trying to protect; too long and Kubernetes SIGKILLs the pod once its grace period expires anyway.
Prerequisites
- KubeMQ server running on
localhost:50000 - Go SDK installed (
go get github.com/kubemq-io/kubemq-go/v2)
Code
// Example: error-handling/graceful-shutdown
//
// Demonstrates graceful shutdown with OS signal handling.
// The client properly drains in-flight operations and closes
// subscriptions when receiving SIGINT or SIGTERM.
//
// Channel: go-error-handling.graceful-shutdown
// Client ID: go-error-handling-graceful-shutdown-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
package main
import (
"context"
"fmt"
"log"
"os"
"os/signal"
"syscall"
"time"
"github.com/kubemq-io/kubemq-go/v2"
)
func main() {
// Create a context that is cancelled on OS signals.
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
client, err := kubemq.NewClient(ctx,
kubemq.WithAddress("localhost", 50000),
kubemq.WithClientId("go-error-handling-graceful-shutdown-client"),
kubemq.WithDrainTimeout(10*time.Second),
kubemq.WithOnClosed(func() {
fmt.Println("[Shutdown] Connection closed")
}),
)
if err != nil {
log.Fatalf("Failed to create client: %v", err)
}
channel := "go-error-handling.graceful-shutdown"
// Start a subscription that runs until shutdown.
sub, err := client.SubscribeToEvents(ctx, channel, "",
kubemq.WithOnEvent(func(event *kubemq.Event) {
fmt.Printf("Received: body=%s\n", event.Body)
}),
kubemq.WithOnError(func(err error) {
log.Println("Subscription error:", err)
}),
)
if err != nil {
log.Fatal(err)
}
fmt.Println("Running... Press Ctrl+C to initiate graceful shutdown")
// Wait for the shutdown signal.
<-ctx.Done()
fmt.Println("\nShutdown signal received, cleaning up...")
// Cancel subscription first.
sub.Unsubscribe()
fmt.Println("Subscription cancelled")
// Close the client, draining in-flight operations.
if err := client.Close(); err != nil {
log.Printf("Close error: %v", err)
}
fmt.Println("Graceful shutdown complete")
}
How It Works
signal.NotifyContextcreates a context that is cancelled when the process receivesSIGINTorSIGTERM, replacing manualos.Signalchannel handling.kubemq.WithDrainTimeout(10 * time.Second)gives in-flight operations up to 10 seconds to complete before the gRPC connection is forcibly closed.kubemq.WithOnClosedregisters a callback that fires when the connection is fully shut down — useful for final logging or metrics flushing.- The explicit
sub.Unsubscribe()beforeclient.Close()ensures the subscription stream is cancelled cleanly before the underlying transport is torn down.
Related
Was this page helpful?