feat(consumers): Auto-restart on single-partition lag

[sentry-junior]

Rust consumers can now fail the Kubernetes health file when one assigned partition stops making useful commit progress, so the pod restarts and Kafka rebalances.

The Snuba healthcheck (--health-check snuba, also selected automatically when the stall timeout is set) adds two per-partition checks on top of the existing file touch:

• Hard stall — in-flight work with no commit advance past consumer.partition_stall_timeout_secs.
• Relative slowdown — over one timeout-sized window, a partition’s commit rate is below consumer.partition_slow_ratio (default 0.25) of the leave-one-out sibling median while still receiving work. Set the ratio to 0 for hard-stall only. Quiet assignments (inclusive median under 50 offsets/s) skip this check.

experimental_healthcheck is renamed to consumer.commit_progress_healthcheck (consumer-level commit or idle progress). The old key is gone; it was unused in sentry-options-automator.

Python consumers are unchanged (arroyo poll-only file touch).

Enable with --health-check-file already set:

consumer.partition_stall_timeout_secs: 300
consumer.partition_slow_ratio: 0.25

Relative slowdown needs at least two active partitions on the same pod. Restart only helps when the owning member is the problem; a hot or broken partition can take its lag with it. During backlog recovery, librdkafka can starve sibling partitions and look like slowdown, which would force extra rebalances.

Requested by PDPM.

--

View Junior Session in Sentry

[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 9abdb3a. Configure here.}

[onewland]

I think this might be a bad idea without more safeguards. It has potential to cause a negative feedback loop.

We've seen in backlog situations before that when a consumer is successfully processing multiple partitions of a topic (after a period of not-processing) where all partitions are quite behind that librdkafa and the broker will queue messages for processing based on partition ordering, meaning some partitions will not make progress for a while until the "front-most" partition drops below a certain threshold where the buffer will receive messages from a wider set. If that happens with this change enabled, you'll force more rebalancing and slow down recovery; a similar pattern to OOMing when we hit a backlog

[sentry]

Bug: The accepted_outcomes_consumer CLI accepts --health-check and --health-check-file arguments, but the Rust implementation ignores them, preventing health checks from functioning as advertised.
_{Severity: HIGH}

Suggested Fix

The Rust implementation for the accepted outcomes consumer in rust_snuba/src/accepted_outcomes_consumer.rs needs to be updated. The accepted_outcomes_consumer_impl function should pass the health_check and health_check_file parameters to the AcceptedOutcomesStrategyFactory. The factory should then include logic to wrap the processing strategy with the appropriate health check wrapper, similar to how it's done in factory_v2.rs for the regular consumer.

Prompt for AI Agent

Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: snuba/cli/accepted_outcomes_consumer.py#L135-L142

Potential issue: The Python CLI in `accepted_outcomes_consumer.py` was updated to accept
`--health-check` options like `commit-progress` and `partition-stall`, along with
`--health-check-file`. However, the corresponding Rust implementation in
`rust_snuba/src/accepted_outcomes_consumer.rs` does not use these parameters, as
indicated by the `_health_check` and `_health_check_file` variable names. As a result,
no health check logic is executed and no health file is created. This will cause
Kubernetes liveness probes to fail for deployments using these new flags, leading to
repeated pod restarts.