fix(produce, consumer): honour acks, and survive leader moves & SSL broker restarts without killing the group

[Argonus]

Scope

This PR bundles two related resilience themes for the 1.1.2 line, both touching the produce/fetch
paths and both needed for a consumer group to survive a broker restart:

1. Produce — the acks option reaches the broker again, and acks: 0 fire-and-forget works.
2. Consumer — a leader move or a broker restart no longer takes the whole group down.

Each is described below. The commit history keeps them separable.

---

1. Produce: honour the acks option, make acks=0 fire-and-forget work

The bug

KafkaEx.API.produce/5 documents a :required_acks option. It never reached the wire.

0.x translated it in the legacy adapter (lib/kafka_ex/new/adapter.ex, acks: produce_request.required_acks). The 1.0 rewrite (573296b) removed that adapter and the translation went with it, while the public docs kept advertising the option. The request builder reads a different key, :acks, so since 1.0 every produce has gone out with acks: -1 regardless of what the caller asked for.

Two more defects sat behind that one, unreachable only because the first hid them:

• acks: 0 crashed the client — byte_size/1 over the :ok returned by the async send path.
• Even with that fixed, the async result didn't match the {{:ok, response}, state} shape the retry layer expects.

What changed

Option resolution. Resolved once in KafkaEx.Client, order :acks > :required_acks > -1, feeding both the wire and the produce telemetry so the two can no longer disagree. :acks is now the canonical spelling; :required_acks is a deprecated alias, honoured until 2.0.

Validation. Anything other than -1/0/1 is rejected with {:error, :invalid_acks}. Unvalidated it reached Kayrock's int16 encoder, where acks: :all (the Java spelling) raised inside handle_call and killed the client, and acks: 65_536 truncated to 0 on the wire while the client waited for a response that would never come.

Fire-and-forget. acks: 0 now takes a dedicated path that skips the retry layer — a resend would duplicate records, since the broker sends nothing to confirm the first one — and returns {:ok, %RecordMetadata{base_offset: nil}}. base_offset widens to non_neg_integer() | nil. Java uses -1 + hasOffset() and brod an unknown-offset marker; nil is used here because this struct has no has_offset?/1 and nil is already what the telemetry code treats as "no offset".

Two failure modes fire-and-forget makes routine:

• A failed async send now closes the broken socket, the way send_sync_request/3 always did. At acks: 0 this is the only feedback channel there is, and discarding it left the next produce reusing a dead socket. (The close event is emitted once, by the client's active-mode error handler — see §2.)
• KafkaEx.Client gains {:tcp_error, _, _} / {:ssl_error, _, _} clauses and a catch-all handle_info/2. Defining any handle_info/2 clause replaces the one use GenServer injects, so an unmatched message crashed the client together with every broker socket and all cluster metadata — and a connection reset, which is exactly how a broker reports a rejected acks: 0 batch, arrives as {:tcp_error, _, :econnreset} ahead of {:tcp_closed, _}.

Upgrade note

Callers who pass neither option are unaffected. Callers who pass one move from the durability they were silently getting to the one they asked for:

caller wrote	before (wire)	after (wire)
nothing	-1	-1
required_acks: 1	-1, all in-sync replicas	1, leader only
required_acks: 0	-1, all in-sync replicas, real offset	0, fire-and-forget, base_offset: nil

Pass acks: -1 explicitly to keep the 1.0/1.1 behaviour. Code doing arithmetic on base_offset after an acks: 0 produce will now raise.

Also worth knowing at acks: 0: {:ok, _} means only that the batch was handed to the socket, and the [:kafka_ex, :produce, :stop] event carries no :offset key — a telemetry handler written as metadata.offset rather than Map.get/2 will raise and be detached.

---

2. Consumer: survive leader moves and broker restarts

The bug

With replication-factor 1 (the AutoMQ default), restarting the broker that leads a partition killed the entire consumer group. The chain:

NOT_LEADER_FOR_PARTITION → metadata parsing drops the leaderless partition
  → select_node → :no_broker → GenConsumer {:stop, :no_broker}
  → ConsumerGroup supervises with max_restarts: 0 → the whole group dies

RF=1 makes the leaderless window wide (no follower to promote), but the same path runs on any leader move.

What changed

• Leaderless partitions stay in metadata. Partitions carrying a leader-move error code (:leader_not_available, :not_leader_for_partition, :not_leader_or_follower, :kafka_storage_error, :replica_not_available) are retained with the leader the broker reported; select_node/2 answers {:error, :leader_not_available} rather than resolving a negative node id to :broker_not_found. This also closes a silent per-key ordering bug: the default partitioner derives the partition count from map_size(partition_leaders), so dropping a leaderless partition remapped keys.
• The fetch loop retries instead of stopping. Any error Retry.fetch_retryable?/1 accepts (a leader move, :no_broker, a closed socket, a timeout) now backs off and retries — unbounded, matching brod, KafkaJS and librdkafka, none of which fail a consumer over a leaderless partition. Jittered exponential backoff (:fetch_retry_base_delay_ms 500 → :fetch_retry_max_delay_ms 5000). The backoff is held as a deadline in state, not a bare GenServer timeout, so an inbound call during a backoff can't cut the wait short. A failed :offset_out_of_range reset rides the same backoff instead of raising a MatchError.
• SSL broker restarts too. A non-atom transport reason (an SSL {:tls_alert, _} tuple) was normalised to :unknown, which the fetch loop treated as fatal — so on TLS clusters the very broker-restart case above still killed the group. Such reasons now normalise to a retryable :transport_error.
• A stuck partition is alertable without reading logs. Because retries never give up, a partition that is genuinely stuck (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 and a [:kafka_ex, :consumer, :partition_unavailable] telemetry event. Mirrors librdkafka's topic.metadata.propagation.max.ms. The consumer keeps retrying.
• :no_broker says why — unknown topic vs. leaderless partition vs. unknown node — logged (at debug; the consumer's throttled retry log is the operator-facing signal). The returned error is unchanged.
• No double connection-close telemetry at acks: 0: the network layer no longer emits its own :send_error close, leaving the active-mode {:tcp_error} handler the single owner of [:kafka_ex, :connection, :close].

What the code review changed (post-review commits)

An adversarial review of the first cut found, and this PR fixes: the :transport_error SSL hole (would have reopened the exact bug on TLS clusters); the duplicate close telemetry; and the amplified :no_broker log. It also confirmed, from the brod / Apache Kafka Java / librdkafka sources, that unknown_topic_or_partition is retried unbounded by all three — so no retry-count cap was added; the 30s time-window signal is the librdkafka-style precedent instead.

Left out on purpose (deferred to 1.2.0)

Returning a concrete error instead of :no_broker (needs an UPGRADING.md entry, not patch/minor material); MetadataLog.missing_topics/2 counting partitions rather than topic keys, and topic-level leaderless retention coupled with it; leader_epoch staleness checking; :kafka_storage_error in fetch_retryable?/1.

---

Tests

• Produce: describe "handle_call/3 - produce acks" and describe "handle_info/2 - socket errors and unknown messages" in client_test.exs, driven through the real handle_call/3/handle_info/2 with NetworkClient stubbed — both spellings and precedence, the effective default, telemetry parity, rejection of unusable values, the fire-and-forget round trip and its error path, and state threading (a correlation_id assertion guarding against a dropped reconnected socket). reliability_test.exs now asserts the real acks: 0 behaviour and sends two async batches before a synchronous request on the same socket, proving the response-less produce leaves nothing to desynchronise framing.
• Consumer: gen_consumer_fetch_retry_test.exs drives the real consume loop through KafkaEx.API.fetch/5 — retries a transient error, retries a non-atom SSL transport error, still stops on a non-retryable error, recovers when the leader returns, backs off on a failed offset reset, serves a call during backoff, and surfaces partition_unavailable once. Plus no_broker_reason_test.exs, the metadata-retention tests, and a retry.ex overflow regression (the old :math.pow backoff raised ArithmeticError past attempt ~1023 — reachable only once retries became unbounded).

Left out of automated runs

Mapping :all/"all" to -1 Java-style (a feature, not a fix), renaming the required_acks telemetry metadata field to acks (a documented contract change), :transactional_id being accepted but inert, and kayrock.md:149 still documenting the 0.x KafkaEx.produce/4 call removed in 1.0.

Verification

Check	Result
mix test.unit	3136 tests, 0 failures, 1 skipped
mix format --check-formatted	clean
mix compile --warnings-as-errors	clean
mix dialyzer	1 warning, identical on an unmodified tree (local MapSet-opaqueness false positive on Elixir 1.19 / OTP 28)
mix credo --strict	could not run fully locally — crashes in Credo.Check.Design.DuplicatedCode on Elixir 1.19; scoped checks on the changed files are clean. Needs CI.
mix test.integration	not run — needs the Docker cluster. Please let CI cover it.

🤖 Generated with Claude Code