Skip to main content

Graceful shutdown

A pipeline drains on SIGTERM (or SIGINT, or a programmatic ShutdownHandle::trigger): it stops taking new data, flushes what it has, commits exactly what was durably written, and exits with a meaningful code. This guide covers the choreography, how to size the one timeout that governs it, and what happens when the deadline is not met. The canonical description is docs/DESIGN.md § Errors, panics, shutdown, health.

The drain choreography​

Shutdown deliberately reuses the same drain path as a full partition revocation — one choreography, tested constantly by every rebalance:

  1. Lanes stop. The controller stops event polling and tells every driver thread to shut down. Each driver stops polling its lanes and drops them — no new data enters the pipeline.
  2. Chains flush. Each driver flushes its operator chain: parked chunks and partial buffers are pushed to the shard queues, then the chain (and with it that thread's ShardQueues clone) is dropped as the thread exits.
  3. Sink batches complete — bounded by checkpoint.drain_timeout. Once every queue sender is gone, the sink workers see their queues close, force-seal partial batches, and get the remaining deadline to finish in-flight writes.
  4. Offsets commit. The controller drains the checkpointer and runs a final synchronous commit + flush_commits — only watermarks covering durably written data, per the at-least-once invariant.

Then everything joins and run returns an ExitReport.

Sizing drain_timeout for Kubernetes​

checkpoint.drain_timeout (default 25s) is the total drain budget. On Kubernetes, SIGTERM starts both this drain and the pod's terminationGracePeriodSeconds clock (default 30s) — after which the kubelet sends SIGKILL. The rule:

[!IMPORTANT] Keep drain_timeout comfortably below terminationGracePeriodSeconds. The gap must absorb steps 1, 2, and 4 plus process teardown, so the final commit always runs before SIGKILL. The defaults (25s vs 30s) encode this; if you raise one, raise the other.

checkpoint:
  drain_timeout: 25s    # keep below terminationGracePeriodSeconds

A SIGKILL mid-drain is still safe for data (nothing uncommitted is lost — it replays), but it forfeits the clean commit, so the restart replays more than it needed to.

Exit codes via ExitReport​

run returns an ExitReport; wire it to the process exit so your orchestrator can tell a drain from a failure:

let report = pipeline.run(source)?;
report.log();                          // info if clean, error if failed
std::process::exit(report.exit_code()); // 0 = Completed, 1 = Failed

ExitState::Completed means the drain finished and the final commit ran — including runs where the sink deadline expired and batches were abandoned (see below; the report's sink_drain field says so). ExitState::Failed carries the failing component and reason; Kubernetes restarts the pod and the data replays. report.ok() converts to a Result when your main prefers ?.

Abandoned batches: replay, not loss​

If the sink cannot flush everything by the deadline — a ClickHouse outage mid-shutdown, say — the remaining batches are abandoned loudly: a log line and a metric, their acknowledgements are failed, and their partitions' watermarks never advance past them. The final commit therefore commits only what was actually written, and the abandoned data replays after restart. That is at-least-once holding under the worst case: an abandon costs duplicate-tolerant replay, never silent loss.

Note the replay caveat: batch dedup tokens do not survive a restart (re-batching changes boundaries), so replayed rows land again — design target tables to tolerate duplicates. See Delivery guarantees and the ClickHouse connector deduplication notes.

Leaked queues: bounded, loud, contained​

The sink workers only enter their drain phase once every ShardQueues clone is dropped. The builder makes the correct drop order structural (queues are lent per chain factory call and die with the driver threads — see the drain contract in Assembling a pipeline). If an assembly nevertheless smuggles a queue clone into long-lived state, shutdown does not hang forever: the drain deadline still fires, and the sink abandons loudly exactly as in an outage — a deadline-bounded loud abandon instead of an unbounded wait. Containment, not absolution: fix the leak, because every shutdown until then replays data unnecessarily.

Testing shutdown​

Disable signal handling and drive the drain yourself — the into_runtime + shutdown_handle pattern in Testing pipelines:

let runtime = pipeline./* ... */.runtime_options(RuntimeOptions {
    handle_signals: false,
    ..RuntimeOptions::default()
}).into_runtime(source)?;
let shutdown = runtime.shutdown_handle();
let join = std::thread::spawn(move || runtime.run());
// drive data, wait for commits...
shutdown.trigger();
let report = join.join().unwrap()?;
assert_eq!(report.exit_code(), 0);
  • Docker and Monitoring — probes, signals, and the metrics that surface abandons and stalled watermarks.
  • Backpressure — why a slow sink shows up long before shutdown does.