Repository: plataformatec/broadway
Pull request #35 (2019-02-13 11:01:59)

Add batcher error handling


Patch:
diff --git a/README.md b/README.md
index 42b537d..641a855 100644
--- a/README.md
+++ b/README.md
@@ -30,62 +30,58 @@ end
 ## Example
 
 ```elixir
-  defmodule MyBroadway do
-    use Broadway
-
-    alias Broadway.Message
-    alias BroadwaySQS.{SQSProducer, ExAwsClient}
-
-    def start_link(_opts) do
-      Broadway.start_link(__MODULE__,
-        name: __MODULE__,
-        producers: [
-          sqs: [
-            module: SQSProducer,
-            arg: [
-              sqs_client: {ExAwsClient, [
-                queue_name: "my_queue",
-              ]}
-            ]
-          ]
-        ],
-        processors: [
-          default: [stages: 50]
-        ],
-        batchers: [
-          s3: [stages: 5, batch_size: 10]
+defmodule MyBroadway do
+  use Broadway
+
+  alias Broadway.Message
+  alias BroadwaySQS.{SQSProducer, ExAwsClient}
+
+  def start_link(_opts) do
+    Broadway.start_link(__MODULE__,
+      name: __MODULE__,
+      producers: [
+        sqs: [
+          module: SQSProducer,
+          arg: [sqs_client: {ExAwsClient, [queue_name: "my_queue"]}]
         ]
-      )
-    end
-
-    def handle_message(_, message, _) do
-      message
-      |> Message.update_data(&process_data/1)
-      |> Message.put_batcher(:s3)
-    end
-
-    def handle_batch(:s3, messages, _, _) do
-      # Send batch of messages to S3
-    end
-
-    defp process_data(data) do
-      # Do some calculations, generate a JSON representation, etc.
-    end
+      ],
+      processors: [
+        default: [stages: 50]
+      ],
+      batchers: [
+        s3: [stages: 5, batch_size: 10, batch_timeout: 1000]
+      ]
+    )
   end
+
+  def handle_message(_processor_name, message, _context) do
+    message
+    |> Message.update_data(&process_data/1)
+    |> Message.put_batcher(:s3)
+  end
+
+  def handle_batch(:s3, messages, _batch_info, _context) do
+    # Send batch of messages to S3
+  end
+
+  defp process_data(data) do
+    # Do some calculations, generate a JSON representation, etc.
+  end
+end
 ```
 
 ## License
 
 Copyright 2019 Plataformatec
 
-  Licensed under the Apache License, Version 2.0 (the "License");
-  you may not use this file except in compliance with the License.
-  You may obtain a copy of the License at
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
 
-      http://www.apache.org/licenses/LICENSE-2.0
+    http://www.apache.org/licenses/LICENSE-2.0
 
-  Unless required by applicable law or agreed to in writing, software
-  distributed under the License is distributed on an "AS IS" BASIS,
-  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-  See the License for the specific language governing permissions and
-  limitations under the License.
\ No newline at end of file
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
diff --git a/lib/broadway/consumer.ex b/lib/broadway/consumer.ex
index 08344e7..76e2fa6 100644
--- a/lib/broadway/consumer.ex
+++ b/lib/broadway/consumer.ex
@@ -3,6 +3,8 @@ defmodule Broadway.Consumer do
   use GenStage
   use Broadway.Subscriber
 
+  require Logger
+
   alias Broadway.{Acknowledger, Message}
   @subscription_options [max_demand: 1, min_demand: 0]
 
@@ -32,9 +34,14 @@ defmodule Broadway.Consumer do
 
     {successful_messages, failed_messages} =
       handle_batch(batcher_key, messages, batch_info, state)
-      |> Enum.split_with(&(&1.status == :ok))
 
-    Acknowledger.ack_messages(successful_messages, failed_messages)
+    try do
+      Acknowledger.ack_messages(successful_messages, failed_messages)
+    catch
+      kind, reason ->
+        Logger.error(Exception.format(kind, reason, System.stacktrace()))
+    end
+
     {:noreply, [], state}
   end
 
@@ -43,10 +50,12 @@ defmodule Broadway.Consumer do
 
     try do
       module.handle_batch(batcher_key, messages, batch_info, context)
-    rescue
-      e ->
-        error_message = Exception.message(e)
-        Enum.map(messages, &Message.failed(&1, error_message))
+      |> Enum.split_with(fn %Message{status: status} -> status == :ok end)
+    catch
+      kind, reason ->
+        Logger.error(Exception.format(kind, reason, System.stacktrace()))
+        failed = "due to an unhandled #{kind}"
+        {[], Enum.map(messages, &Message.failed(&1, failed))}
     end
   end
 end


Test patch:
diff --git a/test/broadway_test.exs b/test/broadway_test.exs
index f69c4d2..5ab5e69 100644
--- a/test/broadway_test.exs
+++ b/test/broadway_test.exs
@@ -381,6 +381,7 @@ defmodule BroadwayTest do
 
       producer = get_producer(broadway_name, :default)
       processor = get_processor(broadway_name, :default)
+      Process.link(Process.whereis(processor))
 
       %{producer: producer, processor: processor}
     end
@@ -854,6 +855,7 @@ defmodule BroadwayTest do
           case message.data do
             :fail -> Message.failed(message, "Failed message")
             :raise -> raise "Error raised"
+            :bad_return -> :oops
             _ -> message
           end
         end)
@@ -878,21 +880,10 @@ defmodule BroadwayTest do
         )
 
       producer = get_producer(broadway_name, :default)
+      consumer = get_consumer(broadway_name, :default)
+      Process.link(Process.whereis(consumer))
 
-      %{producer: producer}
-    end
-
-    test "all messages in the batch are marked as {:failed, reason} when an error is raised",
-         %{producer: producer} do
-      push_messages(producer, [1, 2, 3, :raise])
-
-      assert_receive {:ack, _,
-                      [
-                        %{data: 1, status: {:failed, "Error raised"}},
-                        %{data: 2, status: {:failed, "Error raised"}},
-                        %{data: 3, status: {:failed, "Error raised"}},
-                        %{data: :raise, status: {:failed, "Error raised"}}
-                      ]}
+      %{producer: producer, consumer: consumer}
     end
 
     test "messages are grouped as successful and failed before sent for acknowledgement",
@@ -901,6 +892,58 @@ defmodule BroadwayTest do
 
       assert_receive {:ack, [%{data: 1}, %{data: 4}], [%{data: :fail}, %{data: :fail}]}
     end
+
+    test "all messages in the batch are marked as {:failed, reason} when an error is raised",
+         %{producer: producer, consumer: consumer} do
+      assert capture_log(fn ->
+               push_messages(producer, [1, 2, :raise, 3])
+
+               assert_receive {:ack, _,
+                               [
+                                 %{data: 1, status: {:failed, "due to an unhandled error"}},
+                                 %{data: 2, status: {:failed, "due to an unhandled error"}},
+                                 %{data: :raise, status: {:failed, "due to an unhandled error"}},
+                                 %{data: 3, status: {:failed, "due to an unhandled error"}}
+                               ]}
+             end) =~ "[error] ** (RuntimeError) Error raised"
+
+      refute_received {:EXIT, _, ^consumer}
+    end
+
+    test "all messages in the batch are marked as {:failed, reason} on bad return",
+         %{producer: producer, consumer: consumer} do
+      assert capture_log(fn ->
+               push_messages(producer, [1, 2, :bad_return, 3])
+
+               assert_receive {:ack, _,
+                               [
+                                 %{data: 1, status: {:failed, "due to an unhandled error"}},
+                                 %{data: 2, status: {:failed, "due to an unhandled error"}},
+                                 %{
+                                   data: :bad_return,
+                                   status: {:failed, "due to an unhandled error"}
+                                 },
+                                 %{data: 3, status: {:failed, "due to an unhandled error"}}
+                               ]}
+             end) =~ "[error]"
+
+      refute_received {:EXIT, _, ^consumer}
+    end
+
+    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}}
+
+      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"
+
+      refute_received {:EXIT, _, ^consumer}
+    end
   end
 
   describe "shutdown" do