Graceful Shutdown

Overview

A graceful shutdown stops a KubeMQ client without dropping in-flight messages or leaking server-side subscription state. Killing a process outright, or disposing the 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, 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 cancelling the subscription's CancellationToken, then dispose the client so in-flight operations get a bounded window to finish before the gRPC channel is torn down. CallbackDrainTimeout tells DisposeAsync() how long to wait for in-flight callbacks, cts.Cancel() stops the await foreach iteration as a catchable OperationCanceledException, and Task.WhenAny(subscribeTask, Task.Delay(...)) bounds how long shutdown waits for that iteration to exit.

Gotchas: disposing the client before the subscription task has exited can race the channel teardown — always cancel and wait first. A CallbackDrainTimeout that's too short cuts off the message you were protecting; too long and Kubernetes SIGKILLs the pod once its grace period expires anyway.

Prerequisites

• KubeMQ server running on localhost:50000
• C# SDK installed (dotnet add package KubeMQ.SDK.CSharp)

Code

// KubeMQ .NET SDK — ErrorHandling: Graceful Shutdown
//
// This example demonstrates gracefully shutting down a client that has
// active subscriptions. The client drains in-flight callbacks, cancels
// subscriptions, and disposes the gRPC channel cleanly.
//
// Prerequisites:
//   - KubeMQ server running on localhost:50000
//   - dotnet run

using KubeMQ.Sdk.Client;
using KubeMQ.Sdk.Events;
using System.Text;

var options = new KubeMQClientOptions
{
    Address = "localhost:50000",
    ClientId = "csharp-errorhandling-graceful-shutdown-client",
    CallbackDrainTimeout = TimeSpan.FromSeconds(10),
};

await using var client = new KubeMQClient(options);
await client.ConnectAsync();

Console.WriteLine("Connected to KubeMQ server");

// Start a subscription
var cts = new CancellationTokenSource();
var subscribeTask = Task.Run(async () =>
{
    try
    {
        await foreach (var msg in client.SubscribeToEventsAsync(
            new EventsSubscription { Channel = "csharp-errorhandling.graceful-shutdown" }, cts.Token))
        {
            Console.WriteLine($"Received: {Encoding.UTF8.GetString(msg.Body.Span)}");
        }
    }
    catch (OperationCanceledException)
    {
        Console.WriteLine("Subscription cancelled gracefully.");
    }
});

await Task.Delay(1000);

// Publish a few events
for (var i = 1; i <= 3; i++)
{
    await client.SendEventAsync(new EventMessage
    {
        Channel = "csharp-errorhandling.graceful-shutdown",
        Body = Encoding.UTF8.GetBytes($"Event #{i}"),
    });
}

await Task.Delay(1000);

// Graceful shutdown: cancel subscription, then dispose
Console.WriteLine("Initiating graceful shutdown...");
cts.Cancel();

// Wait briefly for subscription to exit
await Task.WhenAny(subscribeTask, Task.Delay(5000));

// DisposeAsync will drain in-flight callbacks and close the gRPC channel
await client.DisposeAsync();
Console.WriteLine("Client disposed. Shutdown complete.");

Console.WriteLine("Done.");

How It Works

• CallbackDrainTimeout = TimeSpan.FromSeconds(10) tells DisposeAsync() to wait up to 10 seconds for any in-flight subscription callbacks to complete before forcefully closing the channel.
• cts.Cancel() signals the await foreach iteration to stop; the OperationCanceledException is caught inside the background task and logged cleanly.
• Task.WhenAny(subscribeTask, Task.Delay(5000)) provides a bounded wait — if the subscription does not exit within 5 seconds, the program proceeds to dispose anyway.
• The explicit await client.DisposeAsync() call before the end of the await using scope is redundant here but illustrates that DisposeAsync is idempotent and safe to call multiple times.

Related

• C# SDK Reference
• Connection Error
• Reconnection