fix(produce): honour the acks option and make acks=0 fire-and-forget work (#593)

Since the 1.0 rewrite every produce went out at acks: -1: the option was read
under a name the request builder never used. Resolve :acks / :required_acks once
to the wire (:acks canonical, :required_acks a deprecated alias until 2.0),
reject values the int16 encoder cannot carry with {:error, :invalid_acks}, and
accept :all / :any as -1.

acks: 0 now takes a dedicated fire-and-forget path (sent async, never retried,
returns base_offset: nil). A failed async send closes the socket, and the client
gains {:tcp_error} / {:ssl_error} clauses plus a catch-all handle_info/2 so an
unmatched message no longer crashes it together with every broker socket.

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: Argonus

CHANGELOG.md

@@ -1,5 +1,24 @@
 # KafkaEx Changelog
 
+## 1.1.2 (unreleased)
+
+### Changed (Breaking)
+
+* **`acks` / `required_acks` now reaches the broker.** Since 1.0 every produce ran at `acks: -1`
+  regardless of the option. `required_acks: 1` now means leader-only and `required_acks: 0` means
+  fire-and-forget (records can be lost, `base_offset: nil`). Pass `acks: -1` to keep the old
+  behaviour. `:acks` is canonical; `:required_acks` is a deprecated alias (removed in 2.0).
+
+### Fixed
+
+* **The `acks` option is honoured again (regression since 1.0).** Resolved once to the wire;
+  invalid values are rejected with `{:error, :invalid_acks}` instead of crashing Kayrock's int16
+  encoder; `:all` / `:any` map to `-1`.
+* **`acks: 0` fire-and-forget works.** Sent asynchronously, never retried, returns
+  `base_offset: nil`. A failed async send closes the socket, and the client now handles
+  `{:tcp_error}` / `{:ssl_error}` plus a catch-all `handle_info/2` instead of crashing on an
+  unmatched message.
+
 ## 1.1.1 (2026-07-24)
 
 ### Fixed

lib/kafka_ex/api.ex

@@ -810,8 +810,17 @@ defmodule KafkaEx.API do
     * `messages` - List of message maps with `:value`, optional `:key`, `:timestamp`, `:headers`
     * `opts` - Options including:
       * `:api_version` - API version to use. If omitted, resolved via `:api_versions` app-config or broker-negotiated max (`min(broker_max, kayrock_max)`). See CHANGELOG § 3-tier API version resolution.
-      * `:required_acks` - Number of acks required (default: 1)
-      * `:timeout` - Request timeout
+      * `:acks` - Acknowledgements required: `-1` (default) all in-sync replicas, `1` leader only,
+        `0` fire-and-forget. `:all` / `:any` (the Java/librdkafka spelling) are accepted as `-1`.
+        Any other value is rejected with `{:error, :invalid_acks}`.
+        With `0` the broker sends no response at all, so `{:ok, _}` means only that the batch was
+        handed to the socket, the call is never retried, the returned `RecordMetadata` has
+        `base_offset: nil`, and a broker-side rejection (stale leader, oversized batch) arrives as
+        a closed connection — records produced before the client notices are lost.
+      * `:required_acks` - Deprecated alias for `:acks`, honoured only when `:acks` is absent.
+        Will be removed in 2.0.
+      * `:timeout` - How long the broker waits for the required acknowledgements, in ms
+        (default 5000). Only has an effect with `acks: -1`.
       * `:partitioner` - Custom partitioner module (default: configured or `KafkaEx.Producer.Partitioner.Default`)
 
   ## Partitioning

lib/kafka_ex/client/client.ex

@@ -27,6 +27,7 @@ defmodule KafkaEx.Client do
   alias KafkaEx.Cluster.ClusterMetadata
   alias KafkaEx.Messages.Fetch
   alias KafkaEx.Messages.FindCoordinator, as: FindCoordinatorMsg
+  alias KafkaEx.Messages.RecordMetadata
   alias KafkaEx.Support.OptionalDeps
   alias KafkaEx.Support.Retry
 
@@ -395,6 +396,21 @@ defmodule KafkaEx.Client do
     {:noreply, state_out}
   end
 
+  def handle_info({:tcp_error, socket, reason}, state) do
+    {:noreply, close_broker_by_socket(state, socket, :recv_error, reason)}
+  end
+
+  def handle_info({:ssl_error, socket, reason}, state) do
+    {:noreply, close_broker_by_socket(state, socket, :recv_error, reason)}
+  end
+
+  # Defining any handle_info/2 replaces the catch-all use GenServer injects; without one an
+  # unmatched message would crash the client.
+  def handle_info(message, state) do
+    Logger.warning("#{inspect(__MODULE__)} ignoring unexpected message: #{inspect(message)}")
+    {:noreply, state}
+  end
+
   defp update_metadata_with_retry(_state, retries_left) when retries_left <= 0 do
     raise KafkaEx.MetadataUpdateError, attempts: @max_metadata_update_retries
   end
@@ -629,16 +645,14 @@ defmodule KafkaEx.Client do
   end
 
   defp produce_request(topic, partition, messages, opts, state) do
-    message_count = length(messages)
-    required_acks = Keyword.get(opts, :required_acks, 1)
-    client_id = Config.client_id()
-
-    metadata = Telemetry.produce_metadata(topic, partition, client_id, required_acks)
-    start_measurements = %{message_count: message_count}
+    with {:ok, acks} <- resolve_acks(opts) do
+      metadata = Telemetry.produce_metadata(topic, partition, Config.client_id(), acks)
+      start_measurements = %{message_count: length(messages)}
 
-    Telemetry.span([:kafka_ex, :produce], Map.merge(metadata, start_measurements), fn ->
-      do_produce_request(topic, partition, messages, opts, state, metadata)
-    end)
+      Telemetry.span([:kafka_ex, :produce], Map.merge(metadata, start_measurements), fn ->
+        do_produce_request(topic, partition, messages, Keyword.put(opts, :acks, acks), state, metadata)
+      end)
+    end
   end
 
   defp do_produce_request(topic, partition, messages, opts, state, metadata) do
@@ -655,6 +669,16 @@ defmodule KafkaEx.Client do
     end
   end
 
+  # Reject a bad acks value here; unvalidated it reaches Kayrock's int16 encoder and crashes the client.
+  defp resolve_acks(opts) do
+    case Keyword.get(opts, :acks, Keyword.get(opts, :required_acks, -1)) do
+      acks when acks in [-1, 0, 1] -> {:ok, acks}
+      # Java/librdkafka spelling for "all in-sync replicas"; accept it as the -1 the wire carries.
+      acks when acks in [:all, :any] -> {:ok, -1}
+      _ -> {:error, :invalid_acks}
+    end
+  end
+
   defp add_offset_to_metadata(metadata, result) do
     case Map.get(result, :base_offset) do
       nil -> metadata
@@ -847,6 +871,20 @@ defmodule KafkaEx.Client do
     |> handle_request_with_retry(state, @coordinator_max_attempts)
   end
 
+  # acks=0: the broker sends no response, and a resend would duplicate records.
+  defp handle_produce_request(%{acks: 0} = request, node_selector, state) do
+    %NodeSelector{topic: topic, partition: partition} = node_selector
+
+    case network_request(request, node_selector, state) do
+      {{:ok, :no_response}, updated_state} ->
+        {{:ok, RecordMetadata.build(topic: topic, partition: partition, base_offset: nil)}, updated_state}
+
+      {{:error, reason}, updated_state} ->
+        Logger.warning("Fire-and-forget produce to #{topic}/#{partition} failed with #{inspect(reason)}")
+        {{:error, build_transport_error(reason)}, updated_state}
+    end
+  end
+
   defp handle_produce_request(request, node_selector, state) do
     # Produce requests should only retry on leadership errors to avoid duplicates.
     # Timeout errors are NOT safe to retry because the message may have been written
@@ -1394,6 +1432,9 @@ defmodule KafkaEx.Client do
         {{:error, reason}, broker} ->
           {{:error, reason}, 0, broker_to_telemetry_info(broker)}
 
+        {:ok, broker} ->
+          {{:ok, :no_response}, 0, broker_to_telemetry_info(broker)}
+
         {data, broker} when synchronous ->
           {deserialize(data, client_request), byte_size(data), broker_to_telemetry_info(broker)}
 
@@ -1507,10 +1548,10 @@ defmodule KafkaEx.Client do
     {topic_metadata, %{updated_state | allow_auto_topic_creation: allow_auto_topic_creation}}
   end
 
-  defp close_broker_by_socket(state, socket, reason \\ :remote_closed) do
+  defp close_broker_by_socket(state, socket, reason \\ :remote_closed, detail \\ nil) do
     State.update_brokers(state, fn broker ->
       if Broker.has_socket?(broker, socket) do
-        Logger.debug("#{Broker.to_string(broker)} closed connection")
+        log_connection_close(broker, detail)
         # Socket is already closed (received :tcp_closed/:ssl_closed), just emit telemetry
         NetworkClient.close_socket(broker, socket, reason)
         Broker.put_socket(broker, nil)
@@ -1519,4 +1560,9 @@ defmodule KafkaEx.Client do
       end
     end)
   end
+
+  defp log_connection_close(broker, nil), do: Logger.debug("#{Broker.to_string(broker)} closed connection")
+
+  defp log_connection_close(broker, detail),
+    do: Logger.warning("#{Broker.to_string(broker)} closed connection: #{inspect(detail)}")
 end

lib/kafka_ex/messages/record_metadata.ex

@@ -11,7 +11,8 @@ defmodule KafkaEx.Messages.RecordMetadata do
 
     * `:topic` - The topic the messages were produced to
     * `:partition` - The partition the messages were produced to
-    * `:base_offset` - The offset assigned to the first message in the batch
+    * `:base_offset` - The offset assigned to the first message in the batch; `nil` when
+      produced with `acks: 0`, where the broker sends no response
     * `:log_append_time` - The timestamp assigned by the broker (v2+, -1 if not available)
     * `:log_start_offset` - The start offset of the log (v5+)
     * `:throttle_time_ms` - Time in ms the request was throttled (v3+)
@@ -29,7 +30,7 @@ defmodule KafkaEx.Messages.RecordMetadata do
   @type t :: %__MODULE__{
           topic: String.t(),
           partition: non_neg_integer(),
-          base_offset: non_neg_integer(),
+          base_offset: non_neg_integer() | nil,
           log_append_time: integer() | nil,
           log_start_offset: non_neg_integer() | nil,
           throttle_time_ms: non_neg_integer() | nil
@@ -42,7 +43,7 @@ defmodule KafkaEx.Messages.RecordMetadata do
 
     * `:topic` - (required) The topic name
     * `:partition` - (required) The partition number
-    * `:base_offset` - (required) The base offset assigned to the batch
+    * `:base_offset` - (required) The base offset assigned to the batch, `nil` for `acks: 0`
     * `:log_append_time` - The broker-assigned timestamp (-1 if not using LogAppendTime)
     * `:log_start_offset` - The log start offset (v5+)
     * `:throttle_time_ms` - Request throttle time in milliseconds (v3+)
@@ -64,7 +65,7 @@ defmodule KafkaEx.Messages.RecordMetadata do
 
   This is an alias for `base_offset` to match Java API.
   """
-  @spec offset(t()) :: non_neg_integer()
+  @spec offset(t()) :: non_neg_integer() | nil
   def offset(%__MODULE__{base_offset: offset}), do: offset
 
   @doc """

lib/kafka_ex/network/network_client.ex

@@ -102,7 +102,9 @@ defmodule KafkaEx.Network.NetworkClient do
       {_, reason} ->
         broker_str = inspect_broker(broker.host, broker.port)
         Logger.error("Asynchronously sending data to broker #{broker_str} failed with #{inspect(reason)}")
-        reason
+        # A self-close is silent (no {:tcp_closed} to the owner), so emit the close event here.
+        close_socket(broker, socket, :send_error)
+        {:error, reason}
     end
   end
 

test/integration/produce/reliability_test.exs

@@ -17,22 +17,56 @@ defmodule KafkaEx.Integration.Produce.ReliabilityTest do
   end
 
   describe "produce with different acks settings" do
-    test "produce with acks=0 (fire and forget) succeeds", %{client: client} do
+    test "produce with acks=0 (fire and forget) returns no offset", %{client: client} do
       topic_name = generate_random_string()
       _ = create_topic(client, topic_name)
 
+      # A fire-and-forget batch is dropped without a trace while the partition leader is still
+      # being elected, so establish the leader with an acknowledged produce first.
+      {:ok, %RecordMetadata{base_offset: base_offset}} =
+        API.produce(client, topic_name, 0, [%{value: "leader-warmup"}], acks: -1)
+
       messages = [%{value: "acks-0-message"}]
 
-      {:ok, result} = API.produce(client, topic_name, 0, messages, required_acks: 0)
+      {:ok, result} = API.produce(client, topic_name, 0, messages, acks: 0)
+      {:ok, _} = API.produce(client, topic_name, 0, messages, acks: 0)
 
       assert %RecordMetadata{} = result
       assert result.topic == topic_name
       assert result.partition == 0
+      assert result.base_offset == nil
+
+      # Two async sends followed by a synchronous request on the same socket: proves the
+      # response-less produce leaves no bytes behind to desynchronise the next request.
+      wait_for(
+        fn ->
+          {:ok, latest} = API.latest_offset(client, topic_name, 0)
+          latest >= base_offset + 3
+        end,
+        200,
+        25
+      )
+    end
 
-      # Wait for message to arrive and verify via fetch
-      Process.sleep(500)
-      {:ok, latest} = API.latest_offset(client, topic_name, 0)
-      assert latest >= 1
+    test "produce with the deprecated required_acks: 0 alias still fires and forgets", %{client: client} do
+      topic_name = generate_random_string()
+      _ = create_topic(client, topic_name)
+
+      {:ok, %RecordMetadata{base_offset: base_offset}} =
+        API.produce(client, topic_name, 0, [%{value: "leader-warmup"}], acks: -1)
+
+      {:ok, result} = API.produce(client, topic_name, 0, [%{value: "alias-message"}], required_acks: 0)
+
+      assert %RecordMetadata{base_offset: nil} = result
+
+      wait_for(
+        fn ->
+          {:ok, latest} = API.latest_offset(client, topic_name, 0)
+          latest >= base_offset + 2
+        end,
+        200,
+        25
+      )
     end
 
     test "produce with acks=1 (leader only) succeeds", %{client: client} do