OffBroadwayRedisStream - A Redis Stream consumer for Broadway

akash-akya

I’m tinkering around the idea of processing redis-stream messages in Elixir and this is one solution.

off_broadway_redis_stream is a redis-stream consumer for Broadway.

It acts as a consumer within the redis-stream-consumer-group. Supports failover by automatically claiming pending messages of the dead consumer.

Rationale

a way to process redis-stream messages, see more about redis-streams here
for background job processing (among other use-cases). Note that there is a subtle difference in approach when compared to traditional sidekiq like job-processing setup. With Broadway our setup is akin to event-processing rather than “async remote procedure call”, this might not be important most of the time, but separating event and processing helps in use-cases such as “single event - multiple different processor”. Also note that this is not a drop-in replacement to existing solutions yet since it is missing important features such as retry-with-backoff (more on this later).
piggyback on Broadway tooling and approach.

Example:

defmodule MyBroadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module:
          {OffBroadwayRedisStream.Producer,
           [
             redis_client_opts: [host: "localhost"],
             stream: "orders",
             group: "processor-group",
             consumer_name: hostname()
           ]}
      ],
      processors: [
        default: [min_demand: 5, max_demand: 1000]
      ]
    )
  end

  def handle_message(_, message, _) do
    [_id, key_value_list] = message.data
    IO.inspect(key_value_list, label: "Got message")
    message
  end

  @max_attempts 5

  def handle_failed(messages, _) do
    for message <- messages do
      if message.metadata.attempt < @max_attempts do
        Broadway.Message.configure_ack(message, retry: true)
      else
        [id, _] = message.data
        IO.inspect(id, label: "Dropping")
      end
    end
  end

  defp hostname do
    {:ok, host} = :inet.gethostname()
    to_string(host)
  end
end

Acknowledgments & Retries

Both successful and failed messages are acknowledged by default. Use handle_failure callback to handle failures by moving messages to other stream, or to schedule retry using sorted-set or to persist failed jobs etc. Currently, library does not provide any retry strategy precisely because there are multiple solutions which differ depending on use case and it’s left to consumer to implement any such logic (but this might change in future)

But it does provide simple retry strategy using Broadway.Message.configure_ack/2 to handle simple external failures (network error etc). For this user has to explicitly configure to retry the particular failed message. Configured message will be attempted again in next batch.

Message.configure_ack(message, retry: true)

For more information and configuration see module documentation.

Github: off_broadway_redis_stream

5 comments

#library #genstage #redis #broadway #stream

2 1672 5

2024-09-11 16:37:46 UTC

Showing Posts 1 to 5

Show Best Posts
Show All (oldest first)
Show All (newest first)

darnahsan

@akash-akya I am trying to use offbraodway_redis_stream and getting a wierd error starting it. Also any plans to bump redix version to >1.5.0 to allow valkey support ?

getting below error
{:badarg, [{OffBroadwayRedisStream.Producer, :init, 1, [file: ~c"lib/producer.ex", line: 107, error_info: %{cause: {4, :binary, :type, {:already_started, #PID<0.860.0>}}

Stacktrace

** (Mix) Could not start application maverick: Maverick.Application.start(:normal, []) returned an error: shutdown: failed to start child: Maverick.Broadway.WhatsappRedis
    ** (EXIT) an exception was raised:
        ** (MatchError) no match of right hand side value: {:error, {:shutdown, {:failed_to_start_child, #Reference<0.751400248.2202796043.259107>, {:shutdown, {:failed_to_start_child, Maverick.Broadway.WhatsappRedis.Broadway.Producer_1, {:badarg, [{OffBroadwayRedisStream.Producer, :init, 1, [file: ~c"lib/producer.ex", line: 107, error_info: %{cause: {4, :binary, :type, {:already_started, #PID<0.860.0>}}, function: :format_bs_fail, module: :erl_erts_errors}]}, {Broadway.Topology.ProducerStage, :init, 1, [file: ~c"lib/broadway/topology/producer_stage.ex", line: 64]}, {GenStage, :init, 1, [file: ~c"lib/gen_stage.ex", line: 1816]}, {:gen_server, :init_it, 2, [file: ~c"gen_server.erl", line: 980]}, {:gen_server, :init_it, 6, [file: ~c"gen_server.erl", line: 935]}, {:proc_lib, :init_p_do_apply, 3, [file: ~c"proc_lib.erl", line: 241]}]}}}}}}
            (broadway 1.1.0) lib/broadway/topology.ex:58: Broadway.Topology.init/1
            (stdlib 5.2) gen_server.erl:980: :gen_server.init_it/2
            (stdlib 5.2) gen_server.erl:935: :gen_server.init_it/6
            (stdlib 5.2) proc_lib.erl:241: :proc_lib.init_p_do_apply/3

code

 def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module:
          {OffBroadwayRedisStream.Producer,
           [
             redis_client_opts: [
               host: host(),
               port: port(),
               password: password(),
               name: :broadway_whatsapp_redix,
               timeout: 30_000,
               ssl: true,
               socket_opts: [
                 verify: :verify_peer,
                 customize_hostname_check: [
                   match_fun: :public_key.pkix_verify_hostname_match_fun(:https)
                 ],
                 # from CAStore package
                 cacertfile: CAStore.file_path()
               ]
             ],
             stream: topic(),
             group: group(),
             consumer_name: consumer(),
             make_stream: true,
             delete_on_acknowledgment: true
           ]},
        concurrency: producer_concurrency()
      ],
      processors: [
        default: [
          concurrency: processors_concurrency()
        ]
      ]
    )
  end

Post #1

akash-akya

@darnahsan can you check if the redix client name :broadway_whatsapp_redix, is started elsewhere? library starts a new redis client for its operation. You can just drop name param or pass a unique name.

I’ll improve the error handling, the message is misleading.