Full static membership support

lib/broadway_kafka/brod_client.ex

@@ -8,6 +8,8 @@ defmodule BroadwayKafka.BrodClient do
   # We only accept :commit_to_kafka_v2 for now so we hard coded the value
   # to avoid problems in case :brod's default policy changes in the future
   @offset_commit_policy :commit_to_kafka_v2
+  @offset_resolution_attempts 3
+  @offset_resolution_backoff_ms 100
 
   @impl true
   def init(opts) do
@@ -91,15 +93,23 @@ defmodule BroadwayKafka.BrodClient do
   def resolve_offset(topic, partition, current_offset, offset_reset_policy, config) do
     policy = offset_reset_policy_value(offset_reset_policy)
 
+    # This is only for testing.
+    brod = Map.get(config, :brod_module, :brod)
+
     if current_offset == :undefined do
-      lookup_offset(config.hosts, topic, partition, policy, config.client_config)
+      lookup_offset(config.hosts, topic, partition, policy, config.client_config, brod)
     else
-      case :brod.fetch({config.hosts, config.client_config}, topic, partition, current_offset) do
+      result =
+        retry_offset_resolution(fn ->
+          brod.fetch({config.hosts, config.client_config}, topic, partition, current_offset)
+        end)
+
+      case result do
         {:ok, _} ->
           current_offset
 
         {:error, :offset_out_of_range} ->
-          lookup_offset(config.hosts, topic, partition, policy, config.client_config)
+          lookup_offset(config.hosts, topic, partition, policy, config.client_config, brod)
 
         {:error, reason} ->
           raise "cannot resolve offset (hosts=#{inspect(config.hosts)} topic=#{topic} " <>
@@ -121,14 +131,19 @@ defmodule BroadwayKafka.BrodClient do
     ]
   end
 
-  defp lookup_offset(hosts, topic, partition, policy, client_config) do
-    case :brod.resolve_offset(hosts, topic, partition, policy, client_config) do
+  defp lookup_offset(hosts, topic, partition, policy, client_config, brod) do
+    result =
+      retry_offset_resolution(fn ->
+        brod.resolve_offset(hosts, topic, partition, policy, client_config)
+      end)
+
+    case result do
       {:ok, -1} ->
         # `:brod.resolve_offset` returns -1 when asked to resolve a timestamp newer
         # than all the messages in the partition.
         # -1 is not a valid offset you can use with `:brod.fetch` so we need to
         # resolve the latest offset instead
-        lookup_offset(hosts, topic, partition, :latest, client_config)
+        lookup_offset(hosts, topic, partition, :latest, client_config, brod)
 
       {:ok, offset} ->
         offset
@@ -139,6 +154,17 @@ defmodule BroadwayKafka.BrodClient do
     end
   end
 
+  defp retry_offset_resolution(fun, attempts_left \\ @offset_resolution_attempts) do
+    case fun.() do
+      {:error, reason} when reason != :offset_out_of_range and attempts_left > 1 ->
+        Process.sleep(@offset_resolution_backoff_ms)
+        retry_offset_resolution(fun, attempts_left - 1)
+
+      result ->
+        result
+    end
+  end
+
   @impl true
   def update_topics(group_coordinator, topics) do
     if group_coordinator do

lib/broadway_kafka/producer.ex

@@ -87,6 +87,11 @@ defmodule BroadwayKafka.Producer do
       these events are emitted in "span style" when receiving assignments revoked call from consumer group coordinator
       See `:telemetry.span/3`.
 
+    * `[:broadway_kafka, :fenced_instance_id]` - emitted after a producer has
+      stopped consuming because Kafka **fenced** its static group member. The
+      measurement is `:system_time`. The metadata includes `:producer`,
+      `:client_id`, `:group_id`, and `:group_instance_id`.
+
   ## Shared Client Performance
 
   Enabling shared client may drastically decrease performance. Since connection is handled by a single process,
@@ -173,7 +178,9 @@ defmodule BroadwayKafka.Producer do
       shutting_down?: false,
       buffer: :queue.new(),
       max_demand: max_demand,
-      shared_client: config.shared_client
+      shared_client: config.shared_client,
+      client_connected?: false,
+      fenced?: false
     }
 
     {:producer, connect(state)}
@@ -197,6 +204,11 @@ defmodule BroadwayKafka.Producer do
   end
 
   @impl GenStage
+  # When this producer is fenced, we "ignore" demand by always returning zero events.
+  def handle_demand(_incoming_demand, %{fenced?: true} = state) do
+    {:noreply, [], state}
+  end
+
   def handle_demand(incoming_demand, %{demand: demand} = state) do
     maybe_schedule_poll(%{state | demand: demand + incoming_demand}, 0)
   end
@@ -219,13 +231,21 @@ defmodule BroadwayKafka.Producer do
   end
 
   @impl GenStage
+  def handle_cast({:update_topics, _topics}, %{fenced?: true} = state) do
+    {:noreply, [], state}
+  end
+
   def handle_cast({:update_topics, topics}, state) do
     state.client.update_topics(state.group_coordinator, topics)
 
     {:noreply, [], state}
   end
 
   @impl GenStage
+  def handle_info({:poll, _key}, %{fenced?: true} = state) do
+    {:noreply, [], state}
+  end
+
   def handle_info({:poll, key}, %{acks: acks, demand: demand, max_demand: max_demand} = state) do
     # We only poll if:
     #
@@ -253,10 +273,21 @@ defmodule BroadwayKafka.Producer do
     end
   end
 
+  def handle_info(:maybe_schedule_poll, %{fenced?: true} = state) do
+    {:noreply, [], state}
+  end
+
   def handle_info(:maybe_schedule_poll, state) do
     maybe_schedule_poll(%{state | receive_timer: nil}, state.receive_interval)
   end
 
+  def handle_info(
+        {:put_assignments, _group_generation_id, _assignments},
+        %{fenced?: true} = state
+      ) do
+    {:noreply, [], state}
+  end
+
   def handle_info({:put_assignments, group_generation_id, assignments}, state) do
     list =
       Enum.map(assignments, fn assignment ->
@@ -287,19 +318,15 @@ defmodule BroadwayKafka.Producer do
       end)
 
     topics_partitions = Enum.map(list, fn {_, topic, partition, _} -> {topic, partition} end)
-    {broadway_index, processors_allocators, batchers_allocators} = state.allocator_names
-
-    for allocator_name <- processors_allocators do
-      Allocator.allocate(allocator_name, broadway_index, topics_partitions)
-    end
-
-    for allocator_name <- batchers_allocators do
-      Allocator.allocate(allocator_name, broadway_index, topics_partitions)
-    end
+    allocate(state, topics_partitions)
 
     {:noreply, [], %{state | acks: Acknowledger.add(state.acks, list)}}
   end
 
+  def handle_info({:ack, _key, _offsets}, %{fenced?: true} = state) do
+    {:noreply, [], state}
+  end
+
   def handle_info({:ack, key, offsets}, state) do
     %{group_coordinator: group_coordinator, client: client, acks: acks, config: config} = state
     {generation_id, topic, partition} = key
@@ -334,6 +361,13 @@ defmodule BroadwayKafka.Producer do
     {:noreply, [], new_state}
   end
 
+  def handle_info(
+        {:DOWN, _ref, _, {client_id, _}, _reason},
+        %{client_id: client_id, fenced?: true} = state
+      ) do
+    {:noreply, [], %{state | client_connected?: false}}
+  end
+
   def handle_info({:DOWN, _ref, _, {client_id, _}, _reason}, %{client_id: client_id} = state) do
     if coord = state.group_coordinator do
       Process.exit(coord, :shutdown)
@@ -342,7 +376,34 @@ defmodule BroadwayKafka.Producer do
     state = reset_buffer(state)
     schedule_reconnect(state.reconnect_timeout)
 
-    {:noreply, [], %{state | group_coordinator: nil}}
+    {:noreply, [], %{state | group_coordinator: nil, client_connected?: false}}
+  end
+
+  def handle_info(
+        {:DOWN, _ref, _, coord, :fenced_instance_id},
+        %{group_coordinator: coord} = state
+      ) do
+    group_instance_id = get_in(state.config, [:group_config, :group_instance_id])
+
+    Logger.error(
+      "Kafka fenced static group member #{inspect(group_instance_id)} because another live " <>
+        "member uses the same :group_instance_id; this producer will not reconnect"
+    )
+
+    state = pause_fenced_producer(state)
+
+    :telemetry.execute(
+      [:broadway_kafka, :fenced_instance_id],
+      %{system_time: System.system_time()},
+      %{
+        producer: self(),
+        client_id: state.client_id,
+        group_id: state.config[:group_id],
+        group_instance_id: group_instance_id
+      }
+    )
+
+    {:noreply, [], state}
   end
 
   def handle_info({:DOWN, _ref, _, coord, _reason}, %{group_coordinator: coord} = state) do
@@ -356,6 +417,10 @@ defmodule BroadwayKafka.Producer do
     {:noreply, [], state}
   end
 
+  def handle_info(:reconnect, %{fenced?: true} = state) do
+    {:noreply, [], state}
+  end
+
   def handle_info(:reconnect, state) do
     if state.client.connected?(state.client_id) do
       {:noreply, [], connect(state)}
@@ -465,7 +530,7 @@ defmodule BroadwayKafka.Producer do
     %{client: client, group_coordinator: group_coordinator, client_id: client_id} = state
     group_coordinator && Process.exit(group_coordinator, :shutdown)
 
-    if state.shared_client == false do
+    if state.shared_client == false and state.client_connected? do
       client.disconnect(client_id)
     end
 
@@ -576,7 +641,7 @@ defmodule BroadwayKafka.Producer do
 
     case client.setup(self(), client_id, __MODULE__, config) do
       {:ok, coord_pid, _coord_ref} ->
-        %{state | group_coordinator: coord_pid}
+        %{state | group_coordinator: coord_pid, client_connected?: true}
 
       error ->
         raise "Cannot connect to Kafka. Reason #{inspect(error)}"
@@ -673,6 +738,56 @@ defmodule BroadwayKafka.Producer do
     put_in(state.buffer, :queue.new())
   end
 
+  defp pause_fenced_producer(state) do
+    if is_reference(state.receive_timer) do
+      Process.cancel_timer(state.receive_timer)
+    end
+
+    if state.revoke_caller do
+      GenStage.reply(state.revoke_caller, :ok)
+    end
+
+    # We deliberately do NOT clear the partition allocations here. Unlike a
+    # regular revoke, fencing doesn't drain the pipeline first, so messages
+    # already emitted downstream still route through the allocator ETS tables
+    # (via :partition_by). Clearing the tables would make those lookups raise
+    # and crash processors/batchers holding in-flight messages. Since a fenced
+    # producer never gets new assignments, the stale entries are harmless.
+    set_draining_after_revoke!(state.draining_after_revoke_flag, false)
+
+    state = %{
+      state
+      | acks: Acknowledger.new(),
+        buffer: :queue.new(),
+        demand: 0,
+        fenced?: true,
+        group_coordinator: nil,
+        receive_timer: nil,
+        revoke_caller: nil,
+        shutting_down?: true
+    }
+
+    disconnect_private_client(state)
+  end
+
+  defp disconnect_private_client(%{shared_client: true} = state), do: state
+  defp disconnect_private_client(%{client_connected?: false} = state), do: state
+
+  defp disconnect_private_client(state) do
+    :ok = state.client.disconnect(state.client_id)
+    %{state | client_connected?: false}
+  end
+
+  defp allocate(state, topics_partitions) do
+    {broadway_index, processors_allocators, batchers_allocators} = state.allocator_names
+
+    for allocator_name <- processors_allocators ++ batchers_allocators do
+      :ok = Allocator.allocate(allocator_name, broadway_index, topics_partitions)
+    end
+
+    :ok
+  end
+
   defp schedule_reconnect(timeout) do
     Process.send_after(self(), :reconnect, timeout)
   end

lib/broadway_kafka/producer_options.ex

@@ -2,6 +2,17 @@ defmodule BroadwayKafka.ProducerOptions do
   @moduledoc false
 
   group_config_schema = [
+    group_instance_id: [
+      type: {:custom, __MODULE__, :validate_nonempty_string, [:group_instance_id]},
+      doc: """
+      A unique, non-empty string that identifies this consumer group member across restarts.
+      This enables [static group membership](https://kafka.apache.org/39/design/design/#static-membership)
+      and requires a `:brod` with fenced-member support. No released `:brod` has that yet,
+      so this fork depends on [`knocklabs/brod`](https://github.com/knocklabs/brod) (see
+      [kafka4beam/brod#669](https://github.com/kafka4beam/brod/pull/669)). *Available since
+      v0.6.0*.
+      """
+    ],
     offset_commit_interval_seconds: [
       type: :pos_integer,
       default: 5,

mix.exs

@@ -27,7 +27,9 @@ defmodule BroadwayKafka.MixProject do
   defp deps do
     [
       {:broadway, "~> 1.0"},
-      {:brod, "~> 3.16 or ~> 4.0"},
+      # We need our brod fork until https://github.com/kafka4beam/brod/pull/669
+      # (fenced static member support) is merged and released upstream.
+      {:brod, github: "knocklabs/brod", ref: "d16aa8cea37ad9aa8e8591aba0eb2ef814c93e6b"},
       {:nimble_options, "~> 0.3 or ~> 1.0"},
       {:telemetry, "~> 0.4.3 or ~> 1.0"},
       {:ex_doc, ">= 0.19.0", only: :docs}

mix.lock

@@ -1,11 +1,11 @@
 %{
   "broadway": {:hex, :broadway, "1.0.0", "da99ca10aa221a9616ccff8cb8124510b7e063112d4593c3bae50448b37bbc90", [:mix], [{:gen_stage, "~> 1.0", [hex: :gen_stage, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.3.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "b86ebd492f687edc9ad44d0f9e359da70f305b6d090e92a06551cef71ec41324"},
-  "brod": {:hex, :brod, "4.0.0", "a67d46adf057669f331c39b95895723d356cc64f06e7c8bb5b9365056fd4df34", [:rebar3], [{:kafka_protocol, "4.1.5", [hex: :kafka_protocol, repo: "hexpm", optional: false]}], "hexpm", "75a770447928208f20409e9b3fb68462d5cdb34ae960250e5a38d045270300c2"},
-  "crc32cer": {:hex, :crc32cer, "0.1.8", "c6c2275c5fb60a95f4935d414f30b50ee9cfed494081c9b36ebb02edfc2f48db", [:rebar3], [], "hexpm", "251499085482920deb6c9b7aadabf9fb4c432f96add97ab42aee4501e5b6f591"},
+  "brod": {:git, "https://github.com/knocklabs/brod.git", "d16aa8cea37ad9aa8e8591aba0eb2ef814c93e6b", [ref: "d16aa8cea37ad9aa8e8591aba0eb2ef814c93e6b"]},
+  "crc32cer": {:hex, :crc32cer, "1.1.3", "da59f0ddd08f56d9a34186026c0b315cbc02ca390cf8cb15b045f10fb7dcc306", [:rebar3], [], "hexpm", "09c8b567b7be6bf43066185341ec0798eca7dbb66a6ab062ad0e2b25860d7ca1"},
   "earmark_parser": {:hex, :earmark_parser, "1.4.40", "f3534689f6b58f48aa3a9ac850d4f05832654fe257bf0549c08cc290035f70d5", [:mix], [], "hexpm", "cdb34f35892a45325bad21735fadb88033bcb7c4c296a999bde769783f53e46a"},
   "ex_doc": {:hex, :ex_doc, "0.34.1", "9751a0419bc15bc7580c73fde506b17b07f6402a1e5243be9e0f05a68c723368", [:mix], [{:earmark_parser, "~> 1.4.39", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.0", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14 or ~> 1.0", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1 or ~> 1.0", [hex: :makeup_erlang, repo: "hexpm", optional: false]}, {:makeup_html, ">= 0.1.0", [hex: :makeup_html, repo: "hexpm", optional: true]}], "hexpm", "d441f1a86a235f59088978eff870de2e815e290e44a8bd976fe5d64470a4c9d2"},
   "gen_stage": {:hex, :gen_stage, "1.1.1", "78d83b14ca742f4c252770bcdf674d83378ca41579c387c57e2f06d70f596317", [:mix], [], "hexpm", "eb90d2d72609050a66ce42b7d4a69323a60c892a09ead0680d5d8ef16b9a034e"},
-  "kafka_protocol": {:hex, :kafka_protocol, "4.1.5", "d15e64994a8ca99716ab47db4132614359ac1bfa56d6c5b4341fdc1aa4041518", [:rebar3], [{:crc32cer, "0.1.8", [hex: :crc32cer, repo: "hexpm", optional: false]}], "hexpm", "c956c9357fef493b7072a35d0c3e2be02aa5186c804a412d29e62423bb15e5d9"},
+  "kafka_protocol": {:hex, :kafka_protocol, "4.3.4", "a641333ac8732071dc289eca9afb88d1abf6fb7b160b95b9a48b5deffeccf189", [:rebar3], [{:crc32cer, "1.1.3", [hex: :crc32cer, repo: "hexpm", optional: false]}], "hexpm", "2272cee16a7046ebdcaf7440e5cca0ce87110e691c2720e07dfcce2e2e652c04"},
   "makeup": {:hex, :makeup, "1.1.2", "9ba8837913bdf757787e71c1581c21f9d2455f4dd04cfca785c70bbfff1a76a3", [:mix], [{:nimble_parsec, "~> 1.2.2 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "cce1566b81fbcbd21eca8ffe808f33b221f9eee2cbc7a1706fc3da9ff18e6cac"},
   "makeup_elixir": {:hex, :makeup_elixir, "0.16.2", "627e84b8e8bf22e60a2579dad15067c755531fea049ae26ef1020cad58fe9578", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "41193978704763f6bbe6cc2758b84909e62984c7752b3784bd3c218bb341706b"},
   "makeup_erlang": {:hex, :makeup_erlang, "1.0.0", "6f0eff9c9c489f26b69b61440bf1b238d95badae49adac77973cbacae87e3c2e", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "ea7a9307de9d1548d2a72d299058d1fd2339e3d398560a0e46c27dab4891e4d2"},