silviurosu

silviurosu

How to exponentially retry messages if Broadway consumer fails

I have a use case that may not be uncommon.
I read messages from RabbitMQ with Broadway. In handle_message I take messages and process them.
My problem occurs when due to a bug in the code or invalid data the message can not be processed successfully and crashes. Broadway marks the messages as failed. RabbitMQ sends it again and again in an infinite loop making the CPU go to 100% and flooding our bug tracker with messages.
How can I have at least a timeout in the Broadway pipeline until it tries to take another set of messages? I was thinking some exponential retry. I found resubscribe_interval but this is related to producer crashing not consumers.

My solution by now is to put a :timer.sleep() in a rescue clause but I wonder if there is a better way:

defmodule OutcomesConsumer do
  use Broadway

  alias Broadway.Message

  def start_link do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module:
          {BroadwayRabbitMQ.Producer,
           queue: RabbitMQUtils.outcomes_queue(),
           qos: [
             prefetch_count: 6
           ],
           connection: rabbitmq_url()},
        concurrency: 1
      ],
      processors: [
        default: [
          min_demand: 1,
          max_demand: 6,
          concurrency: 2
        ]
      ]
    )
  end

  def handle_message(_, %Message{data: string_data} = message, _) do
    data = Jason.decode!(string_data)
    # process data in another module
    message
  rescue
    e ->
      Logger.error("Error reading outcome from RabbitMQ: #{inspect(e)} ")
      Bugsnag.report(...)

      :timer.sleep(4_000)

      reraise e, __STACKTRACE__
  end
end

First Post!

gregvaughn

gregvaughn

I will possibly be tackling this problem soon. We use SQS as our source, but the ideas are similar. I also solved this in Java in an earlier lifetime. My plan is to set up an extra dead-letter queue (or I might use an S3 bucket, but the concept is similar). The handle_failure callback of Broadway would need to write the failed message to that dead-letter queue and also ACK the original source.

You could then have some secondary producer that reads from the dead-letter queue at a more reasonable interval. Or it could be a manually initiated action to read from there and reprocess those failed messages.

Where Next?

Popular in Questions Top

shahryarjb
Hello, I get Persian date from my client and convert it to normal calendar like this: def jalali_string_to_miladi_english_number(persi...
New
senggen
Erlang/OTP 25 [erts-13.2.2] [source] [64-bit] [smp:8:8] [ds:8:8:10] [async-threads:1] 15:22:35.803 [error] gen_event {lager_file_backend...
New
LegitStack
I’m hoping you guys can give me some general advice and perhaps code examples if you’re feeling up to it. I’m very interested in Elixir,...
New
openscript
Hello! Sorry for this astonishing simple question, but I’m really stuck. I try to set up the intellij-elixir plugin, but I don’t know ho...
New
alice
Hey, Just curious what are the main benefits of Elixir compared to Clojure? When is Elixir more useful than Clojure and vice versa? Th...
New
mathew4509
I have a list say x = ["23gh", "56kh", "97mh"] I would like to pass each element to Val in each iteration. Say, in iteration 1 -------...
New
beno
I will often find my self writing things similar to: case some_value do nil -> something() "" -> something() _ -> someth...
New
Fl4m3Ph03n1x
About me? ( if you have nothing better to do than reading about some random guy in the internet :stuck_out_tongue: ) Hello all, this is ...
New
Qqwy
Original source of discussion: This topic on the Pragmatic Programmers' Functional Web Development with Elixir, OTP, and Phoenix forum. ...
New
jc00ke
Expanding on this topic: https://forum.elixirforum.net/t/map-typespec-question/19217 Let’s say I have a map with required and optional k...
New

Other popular topics Top

lanycrost
Hi everyone! I need implement if…else if…else condition from my elixir code, and anymore of this control flow structures not work proper...
New
vonH
In asking this question I am more interested about the expressiveness of the language itself and less concerned about the availability of...
New
albydarned
Hello all! I am typing this post from my new MacBook Pro with the M1 chip. I’m loving it so far, and will probably use it as my daily dr...
New
aalberti333
As the title describes, I’m trying to run Enum.map() over a list of key/value pairs, where the value is a map. My data looks like this: ...
New
romenigld
I am trying to run a deploy with docker and I successfully runned with this command: docker build -t romenigld/blog-prod . but when I t...
New
vertexbuffer
Hello, can anybody help here..? I have a list of players and I what to delete an element, but every for loop the list is reverting to ori...
New
vac
Hi, I'm quite new in Elixir and I'm trying to format a string to a PEM format. I have the certificate value like MIIDBTCCAe2...... and ...
New
mgjohns61585
Could someone help me? I'm making my first elixir program, number guessing game. I can't figure out how to convert the user's guess from ...
New
qwerescape
Is there a way to get the call stack or stack trace at any point in the code? Not from exceptions, but an expression that returns how the...
New
jay1
Why is it that the mnesia database isn’t the most preferred database for use in Elixir/Phoenix?
New

We're in Beta

About us Mission Statement