View Source KafkaEx.Client.RequestContext (kafka_ex v1.1.0)

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 (Java RetriableException, KafkaJS error.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). The default is &KafkaEx.Support.Retry.data_plane_retryable?/1; produce narrows it to produce_retryable?/1, and the consumer-group operations use their own classifiers. All of them live in KafkaEx.Support.Retry — the single home for error classification, shared with the with_retry loop (stream/consumer).

  • network_timeout is the per-attempt Socket.recv deadline. It is set explicitly by the requests that need a non-generic deadline — fetch (from :max_wait_time), JoinGroup (rebalance_timeout + grace), SyncGroup (session window), Heartbeat (heartbeat_interval) and CreateTopics/DeleteTopics (broker op timeout + grace) — each widening its matching GenServer.call budget in KafkaEx.API so the two cannot mismatch (issue #357). Leave nil for every other request so it falls back to the generic :request_timeout (KafkaEx.Config.request_timeout/0). 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.

t()

Types

@type parser_fn() :: (term() -> {:ok, term()} | {:error, term()})

Projects a deserialized response into a domain result.

@type retryable_fn() :: (term() -> boolean())

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()
}