fix(metadata): keep leaderless partitions in cluster metadata (#595)

Metadata parsing dropped every partition the broker reported with an error code,
so a partition whose leader was moving disappeared from the client's view. The
default partitioner derives its partition count from map_size(partition_leaders),
so that count shrank mid-move and the same key was remapped to a different
partition — silent per-key ordering loss.

Partitions carrying a leader-move error are now retained with the reported leader
(-1 = leader unknown), PartitionInfo carries the broker's error_code, and
ClusterMetadata.select_node/2 answers {:error, :leader_not_available} instead of
resolving a negative node id to :broker_not_found. The no-broker lookup also logs
its cause (unknown topic vs leaderless partition vs unknown node) at debug.

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

Author: Argonus

CHANGELOG.md

@@ -18,6 +18,14 @@
   `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.
+* **A key could silently change partition during a leader move.** Metadata parsing dropped every
+  partition the broker reported with an error, shrinking the count the default partitioner keys
+  off — so the same key could land on a different partition. Partitions carrying a leader-move
+  error are now retained with the reported leader (`-1` = leader unknown),
+  `ClusterMetadata.select_node/2` answers `{:error, :leader_not_available}` for them, and
+  `PartitionInfo` carries the broker's `error_code`.
+* **`:no_broker` says why in the log.** Unknown topic vs. leaderless partition vs. unknown node are
+  now distinguished (logged at debug); the returned error is unchanged.
 
 ## 1.1.1 (2026-07-24)
 

lib/kafka_ex/client/client.ex

@@ -1187,15 +1187,27 @@ defmodule KafkaEx.Client do
         updated_state = state_updater.(state)
 
         case State.select_broker(updated_state, selector) do
-          {:error, _} -> {nil, updated_state}
-          {:ok, broker} -> ensure_broker_connected(broker, updated_state)
+          {:error, reason} ->
+            # Debug, not warning: the consumer's fetch loop retries this lookup and owns the throttled log.
+            Logger.debug("No broker for #{describe_selector(selector)} after metadata refresh: #{inspect(reason)}")
+
+            {nil, updated_state}
+
+          {:ok, broker} ->
+            ensure_broker_connected(broker, updated_state)
         end
 
       {:ok, broker} ->
         ensure_broker_connected(broker, state)
     end
   end
 
+  defp describe_selector(%NodeSelector{strategy: :topic_partition, topic: topic, partition: partition}),
+    do: "#{topic}/#{partition}"
+
+  defp describe_selector(%NodeSelector{strategy: :consumer_group, consumer_group_name: group}),
+    do: "consumer group #{group}"
+
   # Ensures broker is connected, reconnecting if necessary.
   # Returns {broker, updated_state} where broker may have a new socket,
   # or nil if reconnection failed.

lib/kafka_ex/cluster/cluster_metadata.ex

@@ -23,7 +23,8 @@ defmodule KafkaEx.Cluster.ClusterMetadata do
   @typedoc """
   Possible errors given by `select_node/2`
   """
-  @type node_select_error :: :no_such_node | :no_such_topic | :no_such_partition | :no_such_consumer_group
+  @type node_select_error ::
+          :no_such_node | :no_such_topic | :no_such_partition | :no_such_consumer_group | :leader_not_available
 
   @doc """
   List names of topics known by the cluster metadata
@@ -78,6 +79,7 @@ defmodule KafkaEx.Cluster.ClusterMetadata do
       {:ok, %Topic{partition_leaders: partition_leaders}} ->
         case Map.fetch(partition_leaders, partition) do
           :error -> {:error, :no_such_partition}
+          {:ok, node_id} when node_id < 0 -> {:error, :leader_not_available}
           {:ok, node_id} -> {:ok, node_id}
         end
     end

lib/kafka_ex/cluster/partition_info.ex

@@ -8,13 +8,17 @@ defmodule KafkaEx.Cluster.PartitionInfo do
   Java equivalent: `org.apache.kafka.common.PartitionInfo`
   """
 
-  defstruct partition_id: nil, leader: -1, replicas: [], isr: []
+  alias Kayrock.ErrorCode
+
+  # Informational only: node selection keys off `leader` (< 0 = no leader), not this field.
+  defstruct partition_id: nil, leader: -1, replicas: [], isr: [], error_code: :no_error
 
   @type t :: %__MODULE__{
           partition_id: integer(),
           leader: integer(),
           replicas: [integer()],
-          isr: [integer()]
+          isr: [integer()],
+          error_code: atom()
         }
 
   @doc """
@@ -23,7 +27,8 @@ defmodule KafkaEx.Cluster.PartitionInfo do
   ## Parameters
 
   The metadata map should contain:
-    - `:error_code` - Must be 0 for successful parsing
+    - `:error_code` - The broker's partition error code (0 = no error), retained as-is. Unlike the
+      production `parse_partitions/1`, this constructor applies no error-code whitelist
     - `:partition` - The partition number
     - `:leader` - The leader node ID
     - `:replicas` - List of replica node IDs
@@ -32,7 +37,7 @@ defmodule KafkaEx.Cluster.PartitionInfo do
   """
   @spec from_partition_metadata(map()) :: t()
   def from_partition_metadata(%{
-        error_code: 0,
+        error_code: error_code,
         partition_index: partition,
         leader_id: leader,
         replica_nodes: replicas,
@@ -42,7 +47,8 @@ defmodule KafkaEx.Cluster.PartitionInfo do
       partition_id: partition,
       leader: leader,
       replicas: replicas,
-      isr: isr
+      isr: isr,
+      error_code: ErrorCode.code_to_atom(error_code)
     }
   end
 

lib/kafka_ex/cluster/topic.ex

@@ -23,17 +23,14 @@ defmodule KafkaEx.Cluster.Topic do
         partitions: partition_metadata,
         is_internal: is_internal
       }) do
-    partition_leaders =
-      Enum.into(
-        partition_metadata,
-        %{},
-        fn %{error_code: 0, leader_id: leader, partition_index: partition_id} ->
-          {partition_id, leader}
-        end
-      )
-
     partitions = Enum.map(partition_metadata, &PartitionInfo.from_partition_metadata/1)
 
+    # Keep leaderless partitions (leader -1); dropping them shrinks the count the partitioner keys off.
+    partition_leaders =
+      Enum.into(partitions, %{}, fn %PartitionInfo{partition_id: id, leader: leader} ->
+        {id, leader}
+      end)
+
     %__MODULE__{
       name: name,
       partition_leaders: partition_leaders,

lib/kafka_ex/protocol/kayrock/metadata/response_helpers.ex

@@ -85,17 +85,32 @@ defmodule KafkaEx.Protocol.Kayrock.Metadata.ResponseHelpers do
   @spec parse_partitions([map()]) :: [PartitionInfo.t()]
   def parse_partitions(kayrock_partitions) when is_list(kayrock_partitions) do
     kayrock_partitions
-    |> Enum.filter(&(ErrorCode.code_to_atom(&1.error_code) == :no_error))
-    |> Enum.map(fn partition_map ->
+    |> Enum.map(&{ErrorCode.code_to_atom(&1.error_code), &1})
+    |> Enum.filter(fn {error, _} -> usable_partition?(error) end)
+    |> Enum.map(fn {error, partition_map} ->
       %PartitionInfo{
         partition_id: partition_map.partition_index,
         leader: partition_map.leader_id,
         replicas: partition_map.replica_nodes || [],
-        isr: partition_map.isr_nodes || []
+        isr: partition_map.isr_nodes || [],
+        error_code: error
       }
     end)
   end
 
+  # A partition whose leader is moving comes back with an error code and leader
+  # -1. Dropping it makes it indistinguishable from a partition that does not
+  # exist, and shrinks the partition count the partitioner keys off. Keep it and
+  # let node selection report :leader_not_available. Mirrors kpro's
+  # discover_partition_leader/4 and librdkafka's NULL broker delegation.
+  defp usable_partition?(:no_error), do: true
+  defp usable_partition?(:leader_not_available), do: true
+  defp usable_partition?(:replica_not_available), do: true
+  defp usable_partition?(:not_leader_for_partition), do: true
+  defp usable_partition?(:not_leader_or_follower), do: true
+  defp usable_partition?(:kafka_storage_error), do: true
+  defp usable_partition?(_), do: false
+
   @doc """
   Checks if the metadata response contains any errors.
   """

test/kafka_ex/api_test.exs

@@ -308,6 +308,13 @@ defmodule KafkaEx.APITest do
       assert [{:topic_metadata, ["test-topic"]}, {:produce, "test-topic", _partition, ^messages}] = calls
     end
 
+    test "counts a partition whose leader is moving, so keys keep their partition" do
+      settled = %Topic{name: "test-topic", partition_leaders: %{0 => 1, 1 => 2, 2 => 1}}
+      moving = %Topic{name: "test-topic", partition_leaders: %{0 => 1, 1 => -1, 2 => 1}}
+
+      assert partition_for(settled, "user-123") == partition_for(moving, "user-123")
+    end
+
     test "returns error when topic has no partitions (nil partition)" do
       topic_info = %Topic{name: "test-topic", partition_leaders: %{}}
 
@@ -821,4 +828,17 @@ defmodule KafkaEx.APITest do
       assert {:error, :invalid_consumer_group} = KafkaEx.API.set_consumer_group_for_auto_commit(client, nil)
     end
   end
+
+  defp partition_for(topic_info, key) do
+    {:ok, client} =
+      MockClient.start_link(%{
+        topic_metadata: {:ok, [topic_info]},
+        produce: {:ok, %RecordMetadata{topic: topic_info.name, partition: 0, base_offset: 0}}
+      })
+
+    {:ok, _} = KafkaEx.API.produce(client, topic_info.name, nil, [%{key: key, value: "data"}])
+
+    [_, {:produce, _, partition, _}] = MockClient.get_calls(client)
+    partition
+  end
 end

test/kafka_ex/client/no_broker_reason_test.exs

@@ -0,0 +1,48 @@
+defmodule KafkaEx.Client.NoBrokerReasonTest do
+  @moduledoc """
+  `:no_broker` collapses three different causes — unknown topic, unknown
+  partition, unknown node — so the reason has to reach the log, otherwise a
+  leaderless partition is indistinguishable from a deleted topic.
+  """
+  use ExUnit.Case, async: false
+
+  import ExUnit.CaptureLog
+
+  alias KafkaEx.Client
+  alias KafkaEx.Client.NodeSelector
+  alias KafkaEx.Client.State
+  alias KafkaEx.Cluster.ClusterMetadata
+  alias KafkaEx.Cluster.Topic
+
+  defp state_with(topics) do
+    %State{cluster_metadata: %ClusterMetadata{brokers: %{}, topics: topics}}
+  end
+
+  defp fetch_from(state, topic, partition) do
+    request = %Kayrock.Fetch.V0.Request{replica_id: -1, max_wait_time: 1, min_bytes: 1, topics: []}
+    selector = NodeSelector.topic_partition(topic, partition)
+
+    capture_log(fn ->
+      {:reply, reply, _} = Client.handle_call({:network_request, request, selector}, self(), state)
+      send(self(), {:reply, reply})
+    end)
+  end
+
+  test "names the missing partition behind :no_broker" do
+    state = state_with(%{"t" => %Topic{name: "t", partition_leaders: %{0 => 1}}})
+
+    log = fetch_from(state, "t", 7)
+
+    assert_received {:reply, {:error, :no_broker}}
+    assert log =~ "No broker for t/7"
+    assert log =~ ":no_such_partition"
+  end
+
+  test "names the missing topic behind :no_broker" do
+    log = fetch_from(state_with(%{}), "gone", 0)
+
+    assert_received {:reply, {:error, :no_broker}}
+    assert log =~ "No broker for gone/0"
+    assert log =~ ":no_such_topic"
+  end
+end

test/kafka_ex/cluster/cluster_metadata_test.exs

@@ -396,4 +396,23 @@ defmodule KafkaEx.Cluster.ClusterMetadataTest do
       assert ClusterMetadata.known_topics(updated_cluster) == ["topic-one"]
     end
   end
+
+  describe "select_node/2 with a leaderless partition" do
+    test "reports :leader_not_available rather than a bogus node id" do
+      metadata = %ClusterMetadata{
+        brokers: %{1 => %KafkaEx.Cluster.Broker{node_id: 1, host: "h", port: 9092}},
+        topics: %{"t" => %KafkaEx.Cluster.Topic{name: "t", partition_leaders: %{0 => 1, 1 => -1}}}
+      }
+
+      alias KafkaEx.Client.NodeSelector
+
+      assert {:ok, 1} = ClusterMetadata.select_node(metadata, NodeSelector.topic_partition("t", 0))
+
+      assert {:error, :leader_not_available} =
+               ClusterMetadata.select_node(metadata, NodeSelector.topic_partition("t", 1))
+
+      assert {:error, :no_such_partition} =
+               ClusterMetadata.select_node(metadata, NodeSelector.topic_partition("t", 9))
+    end
+  end
 end

test/kafka_ex/cluster/partition_info_test.exs

@@ -39,6 +39,21 @@ defmodule KafkaEx.Cluster.PartitionInfoTest do
 
       assert partition.isr == [21]
     end
+
+    test "maps a zero error_code to :no_error", %{metadata: metadata} do
+      partition = PartitionInfo.from_partition_metadata(metadata)
+
+      assert partition.error_code == :no_error
+    end
+
+    test "retains a leader-move partition instead of crashing, mapping its error_code" do
+      metadata = %{error_code: 5, partition_index: 0, leader_id: -1, replica_nodes: [], isr_nodes: []}
+
+      partition = PartitionInfo.from_partition_metadata(metadata)
+
+      assert partition.leader == -1
+      assert partition.error_code == :leader_not_available
+    end
   end
 
   describe "build/1" do