Graceful Shutdown
Gracefully shut down a KubeMQ Python client, releasing connections and cleaning up resources on exit.
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 cancelling the subscription, then close the client so in-flight operations get a bounded window to finish before the gRPC connection is torn down. A signal.SIGINT handler flips a flag the run loop checks, an AsyncCancellationToken is cancel()-ed and its subscription task.cancel()-ed to stop the async generator, and only then is await client.close() called.
Gotchas: calling client.close() before cancelling the subscription task can raise KubeMQClientClosedError inside the still-running coroutine — always cancel first. Forgetting to await the cancelled task can leave the event loop warning about a task that was never retrieved.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Graceful shutdown — properly close clients and cancel subscriptions."""
from __future__ import annotations
import signal
import asyncio
from kubemq import (
AsyncCancellationToken,
AsyncPubSubClient,
EventMessage,
EventsSubscription,
)
async def main() -> None:
client = AsyncPubSubClient(
address="localhost:50000",
client_id="python-error-handling-graceful-shutdown-client",
)
shutdown_requested = False
def signal_handler(sig: int, frame: object) -> None:
nonlocal shutdown_requested
print("\nShutdown signal received...")
shutdown_requested = True
# Register signal handler for graceful shutdown
signal.signal(signal.SIGINT, signal_handler)
try:
await client.connect()
# Start subscription
token = AsyncCancellationToken()
async def subscriber() -> None:
async for event in client.subscribe_to_events(
subscription=EventsSubscription(
channel="python-error-handling.graceful-shutdown",
on_receive_event_callback=lambda e: None,
on_error_callback=lambda e: print(f"Error: {e}"),
),
cancellation_token=token,
):
print(f"Received: {event.body.decode('utf-8')}")
task = asyncio.create_task(subscriber())
await asyncio.sleep(1)
print("Subscription active. Press Ctrl+C to shutdown gracefully.")
# Send some messages
for i in range(3):
if shutdown_requested:
break
await client.publish_event(
EventMessage(
channel="python-error-handling.graceful-shutdown",
body=f"Message #{i + 1}".encode(),
)
)
await asyncio.sleep(0.5)
await asyncio.sleep(1)
finally:
# Graceful shutdown sequence:
# 1. Cancel all subscriptions
print("Step 1: Cancelling subscriptions...")
token.cancel()
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
# 2. Close the client (drains in-flight operations)
print("Step 2: Closing client...")
await client.close()
print("Shutdown complete")
if __name__ == "__main__":
asyncio.run(main())
How It Works
The client is constructed without async with so shutdown order is explicit. A signal.SIGINT handler sets a flag to break the publish loop. In the finally block, the two-step shutdown is: (1) cancel the AsyncCancellationToken and task.cancel() to stop the subscription coroutine, then (2) await client.close() to drain in-flight gRPC streams and release the connection. Performing these steps out of order — e.g. closing before cancelling the task — can raise KubeMQClientClosedError inside the subscription coroutine.
Related
Was this page helpful?