View Source KafkaEx.Client.RequestContext (kafka_ex v1.0.1)
Invariant inputs for one logical client request, bundled so the retry
functions in KafkaEx.Client (handle_request_with_retry,
handle_request_error) stay small.
The retry loop varies only state, retry_count, and last_error;
everything in this struct is fixed for the life of the request. It merges the
three concerns the loop needs together:
- request identity/payload —
request,node_selector - response projection —
parser_fn(deserialized response -> domain result) - retry policy —
retryable?
Design notes
retryable?is keyed per request, not per error. The reference Kafka clients (JavaRetriableException, KafkaJSerror.retriable, librdkafka) classify retriability on the error, but that cannot work here: the same error atom (e.g.:request_timed_out) is retryable for fetch yet unsafe for produce (duplicate writes). Only produce overrides the default with&KafkaEx.Support.Retry.produce_retryable?/1; everything else inherits always-retry.KafkaEx.Support.Retrycentralizes error-code classification for the separatewith_retryloop (stream/consumer); converging the two is deferred follow-up.network_timeoutis the per-attemptSocket.recvdeadline, set only by fetch (derived from:max_wait_timeinKafkaEx.API.fetch/5, which also widens the matchingGenServer.callbudget). Leavenilfor every other request so it falls back to:sync_timeout. Setting it on a non-fetch request without widening that caller'sGenServer.calltimeout would reintroduce the issue #357 mismatch. It is per-attempt, NOT the total budget.Backoff/jitter between attempts is intentionally absent — the client loop retries immediately (refreshing metadata on leadership errors). If added later, a retry-policy field on this struct is the seam.
Summary
Types
Projects a deserialized response into a domain result.
Decides whether an error should trigger a retry.
Types
Projects a deserialized response into a domain result.
Decides whether an error should trigger a retry.
@type t() :: %KafkaEx.Client.RequestContext{ network_timeout: non_neg_integer() | nil, node_selector: KafkaEx.Client.NodeSelector.t(), parser_fn: parser_fn(), request: struct(), retryable?: retryable_fn() }