feat(consumer): periodically restart the Rust consumer to reconnect to Kafka

[phacops]

Summary

Adds an opt-in mechanism that periodically reconnects the Rust consumer (rust-consumer, the entrypoint every self-hosted consumer service runs) to Kafka. The behavior is gated behind a feature flag and the interval is controlled by a setting, defaulting to 15 minutes.

The reconnect happens in-process: when the interval elapses, the consumer is shut down gracefully and the run loop rebuilds the StreamProcessor (re-establishing the Kafka connection) and resumes consuming — the process is not torn down.

How it's configured

Both knobs are runtime config keys (read from snuba.state), so they can be toggled live without a redeploy — same pattern as the existing quantized_rebalance_consumer_group_delay_secs__<consumer_group> key:

Key	Purpose	Default
kafka_consumer_periodic_restart__<consumer_group>	Feature flag. Enables the behavior when set to a truthy value ("1" / "true").	disabled
kafka_consumer_periodic_restart_interval_secs__<consumer_group>	Setting controlling seconds between reconnects. Must be a positive integer.	900 (15 min)

The interval is only honored when the feature flag is enabled; an invalid/missing interval falls back to the 15‑minute default.

How it works

consumer_impl was reworked so that process-wide, one-time setup (logging, Sentry, metrics, procspawn, the Ctrl-C handler) runs once, and the consumer is built and run inside a restart loop:

• A background timer (spawned per run when the flag is enabled) sleeps for the configured interval, sets a restart_triggered flag, and signals a graceful shutdown.
• When processor.run() returns, the loop checks the flags:
  • periodic restart (restart_triggered) → rebuild the consumer and continue (reconnect, stay alive);
  • Ctrl-C (shutdown_requested) → exit the loop / process;
  • any other graceful termination (e.g. --stop-at-timestamp) → exit the loop.
• Because the ProcessorHandle is recreated on each restart but the Ctrl-C handler can only be installed once, the handler signals the live consumer through a shared Mutex<Option<ProcessorHandle>>.
• The reconnect honors the same delay_kafka_rebalance quantization the Ctrl-C path uses (on both leave and rejoin), so groups that also enable quantized rebalancing don't incur extra Kafka rebalances.

Emits a periodic_restart counter when a reconnect is signaled and a consumer_reconnect counter when the loop actually rebuilds the consumer.

Changes

• New rust_snuba/src/auto_restart.rs — get_restart_interval(consumer_group) -> Option<Duration> reads the two runtime config keys and returns the interval when enabled (or None when disabled). Includes unit tests.
• rust_snuba/src/consumer.rs — consumer_impl restructured into a one-time-setup + restart-loop shape as described above.

Testing

• cargo check --lib ✅
• cargo clippy --lib ✅ (no new warnings)
• rustfmt --check ✅
• cargo test --lib ✅ — all runnable tests pass (the 4 failures in my sandbox require ClickHouse/Redis/Python-state services and fail identically on the base commit; they pass in CI).

Notes

• Default-off: existing consumers are unaffected until the flag is set for their consumer group.
• A periodic reconnect triggers a Kafka group rebalance (the consumer leaves and rejoins). For groups that also use quantized rebalancing, both the leave and the rejoin are quantized to reduce churn.

🤖 Generated with Claude Code

https://claude.ai/code/session_01JXUY9wZfDEGZRMuBUM1AKo

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 69c2a00. Configure here.}