View Source KafkaEx.Support.Retry (kafka_ex v1.0.1)
Unified retry logic with exponential backoff for KafkaEx operations.
This module provides:
- Exponential backoff calculation with optional cap
- Generic retry wrapper function
- Error classifiers for common Kafka error patterns
Usage
# Simple retry with defaults (3 attempts, 100ms base delay)
Retry.with_retry(fn -> some_operation() end)
# Custom retry configuration
Retry.with_retry(
fn -> some_operation() end,
max_attempts: 5,
base_delay_ms: 200,
max_delay_ms: 5000,
retryable?: &Retry.transient_error?/1
)
# Just calculate backoff delay
delay = Retry.backoff_delay(attempt, 100, 5000)Summary
Functions
Calculate exponential backoff delay.
Check if a commit error is fatal and requires rejoining the consumer group.
Check if error is safe to retry for commit operations.
Check if a commit error is terminal: the consumer must stop, NOT rejoin, NOT retry.
Check if error is safe to retry for consumer group operations.
Check if error is a coordinator-related error that may resolve with retry.
Check if a Heartbeat error is safe to retry at the client layer.
Check if the error is safe to retry for join group operations.
Check if error is a leadership-related error requiring metadata refresh.
Check if error is safe to retry for produce operations.
Check if a SyncGroup error is safe to retry at the client layer.
Check if error is a transient error that may resolve with retry.
Execute a function with retry and exponential backoff.
Types
@type retry_opts() :: [ max_attempts: non_neg_integer(), base_delay_ms: non_neg_integer(), max_delay_ms: non_neg_integer() | :infinity, retryable?: (error() -> boolean()), on_retry: (error(), non_neg_integer(), non_neg_integer() -> :ok) | nil ]
Functions
@spec backoff_delay( non_neg_integer(), non_neg_integer(), non_neg_integer() | :infinity ) :: non_neg_integer()
Calculate exponential backoff delay.
Parameters
attempt- Zero-based attempt number (0 = first retry)base_ms- Base delay in millisecondsmax_ms- Maximum delay cap (default::infinity)
Examples
iex> Retry.backoff_delay(0, 100)
100
iex> Retry.backoff_delay(1, 100)
200
iex> Retry.backoff_delay(2, 100)
400
iex> Retry.backoff_delay(10, 100, 5000)
5000Check if a commit error is fatal and requires rejoining the consumer group.
These errors cannot be fixed by retry — the consumer's generation or member_id is no longer valid. Callers must propagate these to the group manager to trigger a rebalance.
Follows Java ConsumerCoordinator.OffsetCommitResponseHandler, brod stabilize/3, kafka-python _handle_offset_commit_response patterns.
Note: :rebalance_in_progress is intentionally NOT in this set — it is
retryable; see commit_retryable?/1. :fenced_instance_id and several
authorization/size errors are classified as commit_terminal_error?/1.
Check if error is safe to retry for commit operations.
Commits are idempotent so we can safely retry on transient errors.
:rebalance_in_progress is also retryable: per Java
ConsumerCoordinator.OffsetCommitResponseHandler, the broker signals that
a rebalance is in flight — we let the heartbeat path drive the eventual
rebalance and simply let the commit retry succeed once the rebalance
completes. Eager rejoin on this code wastes a round trip and can double-
rebalance.
:unstable_offset_commit (KIP-447) — the broker is waiting on an
in-progress transaction; retry is the correct handling.
Check if a commit error is terminal: the consumer must stop, NOT rejoin, NOT retry.
:fenced_instance_id(KIP-345) — another member has taken thisgroup.instance.id. Rejoining would either split-brain (two consumers claim the same static slot) or be fenced again. Java raisesFencedInstanceIdException; librdkafka and brod also treat as non-recoverable.:group_authorization_failed/:topic_authorization_failed— the credentials don't grant access to the group or topic. Java raisesGroupAuthorizationException/TopicAuthorizationExceptionout ofpoll(). Retrying or rejoining will not help.:offset_metadata_too_large/:invalid_commit_offset_size— malformed commit payload. Java treats both as non-retriableKafkaException. Retrying would hot-loop on a caller bug.
Check if error is safe to retry for consumer group operations.
Following Java client pattern (KAFKA-6829): includes coordinator errors, transient errors, and UNKNOWN_TOPIC_OR_PARTITION.
Check if error is a coordinator-related error that may resolve with retry.
These errors typically occur during consumer group operations when the coordinator is unavailable or changing.
Check if a Heartbeat error is safe to retry at the client layer.
The heartbeat process stops on any error and lets the manager react
(Heartbeat.handle_info/2), so the only retry is here at the client. Its
value is absorbing a transient blip (timeout / coordinator hiccup) so a
single failed heartbeat does not escalate into a full rejoin + group-wide
rebalance — hence default-retry for unclassified errors.
The broker's "rejoin now" / identity signals — :rebalance_in_progress,
:unknown_member_id and :illegal_generation — are mapped by the heartbeat
process to a shutdown the manager turns into a rejoin; retrying them at the
client only delays the inevitable rejoin, so they are not retryable. The
terminal codes :fenced_instance_id (another member holds this
group.instance.id) and :group_authorization_failed (no access to the
group) cannot be helped by retry either, so they are not retryable.
Check if the error is safe to retry for join group operations.
KIP-394: JoinGroup V4+ requires two-step join. First attempt with empty member_id returns :member_id_required with an assigned member_id. Then another request needs to be performed with that member_id.
We do not want to retry the same request if the error is
:member_id_required, but rather swap in the assigned member_id.
Check if error is a leadership-related error requiring metadata refresh.
These errors indicate the client has stale partition leadership information. Safe to retry for ALL request types including produce.
Check if error is safe to retry for produce operations.
Produce retries are only safe for leadership errors where we know the message wasn't written. Timeout errors are NOT safe because the message may have been written but the response lost.
Note: For truly idempotent produces, enable enable.idempotence=true
on the Kafka producer (requires Kafka 0.11+).
Check if a SyncGroup error is safe to retry at the client layer.
:rebalance_in_progress must not be retried at the client layer. A SyncGroup
during a rebalance can only succeed after rejoining (the generation is
changing), so the error must bubble to the consumer-group manager, whose
sync/2 rejoins on it. Blindly re-sending the same SyncGroup wastes round
trips and is the window in which a recv-timeout gets mapped to :unknown
(which would otherwise crash the manager). Other errors keep the default
retry behavior.
Contrast commit_retryable?/1, where retrying :rebalance_in_progress is
correct — commits are idempotent and need no rejoin.
:illegal_generation / :unknown_member_id / :fenced_instance_id /
:group_authorization_failed are also not retryable: the same SyncGroup can
only fail the same way. They must bubble so the consumer-group manager rejoins
(resetting identity per the reason) or, for the terminal codes
(:fenced_instance_id, :group_authorization_failed), stops.
Check if error is a transient error that may resolve with retry.
Includes timeouts, parse errors, connection issues, and coordinator errors.
@spec with_retry((-> retry_result()), retry_opts()) :: retry_result()
Execute a function with retry and exponential backoff.
Options
:max_attempts- Maximum number of attempts, including the first (default: 3):base_delay_ms- Base delay for exponential backoff (default: 100):max_delay_ms- Maximum delay cap (default::infinity):retryable?- Function to determine if error is retryable (default: always true):on_retry- Optional callback(error, attempt, delay) -> :okfor logging
Returns
{:ok, result}on success{:error, last_error}after all retries exhausted or non-retryable error
Examples
# Retry any error up to 3 times
Retry.with_retry(fn -> fetch_data() end)
# Only retry specific errors
Retry.with_retry(
fn -> commit_offset() end,
retryable?: &Retry.coordinator_error?/1
)