fix(consumer): retry fetch failures instead of taking the group down

[Argonus]

Last of three related fixes. master already carries the produce (#593) and leaderless-metadata (#595) work; this PR is the runtime piece that makes a leader move survivable. The metadata fix keeps a transiently-leaderless partition in the cluster model; this fix makes the consumer back off and retry it instead of stopping.

Summary

A fetch error used to stop the consumer: handle_info(:timeout) returned {:stop, reason, state} on any {:error, reason}. Because KafkaEx.Consumer.ConsumerGroup supervises with max_restarts: 0, one transient fetch failure — a leader move, a closed socket, an SSL alert — took the whole group down. This branch makes the consumer's fetch loop back off and retry transient/leadership errors, and only stop on errors that are genuinely fatal.

Changes

gen_consumer.ex — retry loop

• handle_info(:timeout) now honours a backoff deadline held in state (fetch_retry_at) rather than stopping. It has to be a state deadline, not a bare GenServer timeout, because any inbound message re-arms the callback at timeout 0.
• handle_fetch_error/2: if Retry.fetch_retryable?/1 accepts the error, compute a jittered exponential backoff (:fetch_retry_base_delay_ms 250 → :fetch_retry_max_delay_ms 5000, both app-env settable) and reschedule; otherwise {:stop, reason, state}. Retries are unbounded, matching brod / KafkaJS / librdkafka.
• A stuck partition (not mid-move — a deleted topic, a leader that never returns) is surfaced once, after :fetch_unavailable_warn_ms (default 30000) of continuous failure, as a Logger.error plus a [:kafka_ex, :consumer, :partition_unavailable] telemetry event. The consumer keeps retrying. Mirrors librdkafka's topic.metadata.propagation.max.ms.
• Retry logging is loud for the first few attempts, then ~once a minute.
• handle_offset_out_of_range now returns {:ok, state} / {:error, reason} instead of pattern-matching the reset offset with =. The reset needs a live leader too, so it fails exactly during a leader move — this hands the error back to the backoff path instead of crashing with a MatchError.

retry.ex

• New fetch_retryable?/1 = transient_error?/1 or leadership_error?/1 — the consumer fetch-loop retriability predicate.
• transient_error?(:transport_error) → true.
• backoff_delay/3 uses an integer bit-shift with a clamped exponent instead of :math.pow, which overflows to ArithmeticError past attempt 1023 (reachable now that fetch retries are unbounded).

client.ex

• build_transport_error/1 maps a non-atom reason (an SSL {:tls_alert, _} tuple) to a retryable :transport_error instead of :unknown, which the fetch loop treated as fatal — this is what took TLS clusters' groups down.

telemetry.ex

• New [:kafka_ex, :consumer, :partition_unavailable] event + emit_partition_unavailable/5, added to the consumer-group event list and moduledoc.

Tests

• gen_consumer_fetch_retry_test.exs: retry-not-stop on :no_broker and on an SSL-alert transport error (asserted via process {:DOWN} monitoring), a call served mid-backoff not stranding the consumer, still-stop on a non-retryable error, the once-only unavailable telemetry, recovery once the leader returns, and backoff on an unreachable offset reset.
• retry_test.exs, telemetry_test.exs, mock_client.ex updated.

Worth a closer look

• Retries are unbounded. A permanently dead partition retries forever, flagged after 30s via telemetry. Deliberate, matching brod / KafkaJS / librdkafka, documented in the moduledoc — worth a conscious accept.
• Startup is still not resilient. :none auto_offset_reset still raises on offset-out-of-range, and load_offsets/1 still raises if the starting offset can't be read, so a consumer that starts while its leader is moving still fails to start. Only the running consumer is now resilient. Documented as an intentional limit.

🤖 Generated with Claude Code