Repository: plataformatec/broadway
Pull request #43 (2019-02-14 16:34:38)

Change acknowledger to be a triplet

josevalim: Two small changess and we should be good to go!

Patch:
diff --git a/lib/broadway/acknowledger.ex b/lib/broadway/acknowledger.ex
index 104d412..98d4356 100644
--- a/lib/broadway/acknowledger.ex
+++ b/lib/broadway/acknowledger.ex
@@ -20,6 +20,14 @@ defmodule Broadway.Acknowledger do
   @doc """
   Invoked to acknowledge successful and failed messages.
 
+    * `ack_ref` is a term that uniquely identifies how messages
+      should be grouped and sent for acknowledgement. Imagine
+      you have a scenario where messages are coming from
+      different producers. Broadway will use this information
+      to correctly identify the acknowledger and pass it among
+      with the messages so you can property communicate with
+      the source of the data for acknowledgement.
+
     * `successful` is the list of messages that were
       successfully processed and published.
 
@@ -27,7 +35,7 @@ defmodule Broadway.Acknowledger do
       could not be processed or published.
 
   """
-  @callback ack(successful :: [Message.t()], failed :: [Message.t()]) :: no_return
+  @callback ack(ack_ref :: reference, successful :: [Message.t()], failed :: [Message.t()]) :: :ok
 
   @doc """
   Acknowledges successful and failed messages grouped by acknowledger.
@@ -41,16 +49,17 @@ defmodule Broadway.Acknowledger do
   end
 
   defp group_by_acknowledger(ackers, messages, key) do
-    Enum.reduce(messages, ackers, fn %{acknowledger: {acknowledger, _}} = msg, acc ->
-      pdict_key = {acknowledger, key}
-      Process.put(pdict_key, [msg | Process.get({acknowledger, key}, [])])
-      Map.put(acc, acknowledger, true)
+    Enum.reduce(messages, ackers, fn %{acknowledger: {acknowledger, ack_ref, _}} = msg, acc ->
+      ack_info = {acknowledger, ack_ref}
+      pdict_key = {ack_info, key}
+      Process.put(pdict_key, [msg | Process.get(pdict_key, [])])
+      Map.put(acc, ack_info, true)
     end)
   end
 
-  defp call_ack({acknowledger, true}) do
-    successful = Process.delete({acknowledger, :successful}) || []
-    failed = Process.delete({acknowledger, :failed}) || []
-    acknowledger.ack(Enum.reverse(successful), Enum.reverse(failed))
+  defp call_ack({{acknowledger, ack_ref} = ack_info, true}) do
+    successful = Process.delete({ack_info, :successful}) || []
+    failed = Process.delete({ack_info, :failed}) || []
+    acknowledger.ack(ack_ref, Enum.reverse(successful), Enum.reverse(failed))
   end
 end
diff --git a/lib/broadway/message.ex b/lib/broadway/message.ex
index a57e0a0..f351a50 100644
--- a/lib/broadway/message.ex
+++ b/lib/broadway/message.ex
@@ -16,7 +16,7 @@ defmodule Broadway.Message do
 
   @type t :: %Message{
           data: any,
-          acknowledger: {module, data :: any},
+          acknowledger: {module, ack_ref :: any, data :: any},
           batcher: atom,
           partition: term,
           status: :ok | {:failed, reason :: binary}


Test patch:
diff --git a/test/broadway_test.exs b/test/broadway_test.exs
index 0cc4110..f15a088 100644
--- a/test/broadway_test.exs
+++ b/test/broadway_test.exs
@@ -9,13 +9,13 @@ defmodule BroadwayTest do
   defmodule Acker do
     @behaviour Broadway.Acknowledger
 
-    def ack(successful, failed) do
+    def ack(_ack_ref, successful, failed) do
       test_pid =
         case {successful, failed} do
-          {[%Message{acknowledger: {_, %{test_pid: pid}}} | _], _failed} ->
+          {[%Message{acknowledger: {_, _ack_ref, %{test_pid: pid}}} | _], _failed} ->
             pid
 
-          {_successful, [%Message{acknowledger: {_, %{test_pid: pid}}} | _]} ->
+          {_successful, [%Message{acknowledger: {_, _ack_ref, %{test_pid: pid}}} | _]} ->
             pid
         end
 
@@ -96,7 +96,7 @@ defmodule BroadwayTest do
 
       %Message{
         data: "#{event} transformed",
-        acknowledger: {Acker, %{id: event, test_pid: test_pid}}
+        acknowledger: {Acker, :ack_ref, %{id: event, test_pid: test_pid}}
       }
     end
   end
@@ -264,8 +264,8 @@ defmodule BroadwayTest do
       )
 
     Broadway.push_messages(pid, [
-      %Message{data: 1, acknowledger: {__MODULE__, 1}},
-      %Message{data: 3, acknowledger: {__MODULE__, 3}}
+      %Message{data: 1, acknowledger: {__MODULE__, :ack_ref, 1}},
+      %Message{data: 3, acknowledger: {__MODULE__, :ack_ref, 3}}
     ])
 
     assert_receive {:message_handled, 1}
@@ -378,12 +378,12 @@ defmodule BroadwayTest do
     test "processors do not crash on bad acknowledger",
          %{producer: producer, processor: processor} do
       [one, raise, four] = wrap_messages([1, :raise, 4])
-      raise = %{raise | acknowledger: {Unknown, :ok}}
+      raise = %{raise | acknowledger: {Unknown, :ack_ref, :ok}}
 
       assert capture_log(fn ->
                Producer.push_messages(producer, [one, raise, four])
                assert_receive {:ack, [%{data: 1}, %{data: 4}], []}
-             end) =~ "[error] ** (UndefinedFunctionError) function Unknown.ack/2 is undefined"
+             end) =~ "[error] ** (UndefinedFunctionError) function Unknown.ack/3 is undefined"
 
       refute_received {:EXIT, _, ^processor}
     end
@@ -825,7 +825,7 @@ defmodule BroadwayTest do
 
       broadway_name = new_unique_name()
 
-      {:ok, _pid} =
+      {:ok, broadway} =
         Broadway.start_link(CustomHandlers,
           name: broadway_name,
           context: context,
@@ -840,14 +840,20 @@ defmodule BroadwayTest do
       consumer = get_consumer(broadway_name, :default)
       Process.link(Process.whereis(consumer))
 
-      %{producer: producer, consumer: consumer}
+      %{broadway: broadway, producer: producer, consumer: consumer}
     end
 
-    test "messages are grouped as successful and failed before sent for acknowledgement",
-         %{producer: producer} do
-      push_messages(producer, [1, :fail, :fail, 4])
+    test "messages are grouped by ack_ref + status (successful or failed) before sent for acknowledgement",
+         %{broadway: broadway} do
+      Broadway.push_messages(broadway, [
+        %Message{data: 1, acknowledger: {Acker, :ack_ref_1, %{test_pid: self(), data_id: 1}}},
+        %Message{data: :fail, acknowledger: {Acker, :ack_ref_2, %{test_pid: self(), data_id: 2}}},
+        %Message{data: :fail, acknowledger: {Acker, :ack_ref_1, %{test_pid: self(), data_id: 3}}},
+        %Message{data: 4, acknowledger: {Acker, :ack_ref_2, %{test_pid: self(), data_id: 4}}}
+      ])
 
-      assert_receive {:ack, [%{data: 1}, %{data: 4}], [%{data: :fail}, %{data: :fail}]}
+      assert_receive {:ack, [%{data: 1}], [%{data: :fail, acknowledger: {_, _, %{data_id: 3}}}]}
+      assert_receive {:ack, [%{data: 4}], [%{data: :fail, acknowledger: {_, _, %{data_id: 2}}}]}
     end
 
     test "all messages in the batch are marked as {:failed, reason} when an error is raised",
@@ -890,14 +896,13 @@ defmodule BroadwayTest do
     test "consumers do not crash on bad acknowledger",
          %{producer: producer, consumer: consumer} do
       [one, two, raise, three] = wrap_messages([1, 2, :raise, 3])
-      raise = %{raise | acknowledger: {Unknown, :ok}}
+      raise = %{raise | acknowledger: {Unknown, :ack_ref, :ok}}
 
       assert capture_log(fn ->
                Producer.push_messages(producer, [one, two, raise, three])
                push_messages(producer, [1, 2, 3, 4])
                assert_receive {:ack, [%{data: 1}, %{data: 2}, %{data: 3}, %{data: 4}], []}
-             end) =~
-               "[error] ** (UndefinedFunctionError) function Unknown.ack/2 is undefined"
+             end) =~ "[error] ** (UndefinedFunctionError) function Unknown.ack/3 is undefined"
 
       refute_received {:EXIT, _, ^consumer}
     end
@@ -1018,7 +1023,7 @@ defmodule BroadwayTest do
 
   defp wrap_messages(list) do
     Enum.map(list, fn data ->
-      %Message{data: data, acknowledger: {Acker, %{id: data, test_pid: self()}}}
+      %Message{data: data, acknowledger: {Acker, :ack_ref, %{id: data, test_pid: self()}}}
     end)
   end
 end