Broadway is a concurrent, multi-stage tool for building data ingestion and data processing pipelines.

It allows developers to consume data efficiently from different sources, such as Amazon SQS, Apache Kafka, Google Cloud PubSub, RabbitMQ and others.

Built-in features

• Back-pressure - by relying on GenStage, we only get the amount of events necessary from upstream sources, never flooding the pipeline.

• Automatic acknowledgements - Broadway automatically acknowledges messages at the end of the pipeline or in case of errors.

• Batching - Broadway provides built-in batching, allowing you to group messages either by size and/or by time. This is important in systems such as Amazon SQS, where batching is the most efficient way to consume messages, both in terms of time and cost.

• Fault tolerance with minimal data loss - Broadway pipelines are carefully designed to minimize data loss. Producers are isolated from the rest of the pipeline and automatically resubscribed to in case of failures. On the other hand, user callbacks are stateless, allowing us to handle any errors locally. Finally, in face of any unforeseen bug, we restart only downstream components, avoiding data loss.

• Graceful shutdown - Broadway integrates with the VM to provide graceful shutdown. By starting Broadway as part of your supervision tree, it will guarantee all events are flushed once the VM shuts down.

• Built-in testing - Broadway ships with a built-in test API, making it easy to push test messages through the pipeline and making sure the event was properly processed.

• Custom failure handling - Broadway provides a handle_failed/2 callback where developers can outline custom code to handle errors. For example, if they want to move messages to another queue for further processing.

• Dynamic batching - Broadway allows developers to batch messages based on custom criteria. For example, if your pipeline needs to build batches based on the user_id, email address, etc, it can be done by calling Broadway.Message.put_batch_key/2.

• Ordering and Partitioning - Broadway allows developers to partition messages across workers, guaranteeing messages within the same partition are processed in order. For example, if you want to guarantee all events tied to a given user_id are processed in order and not concurrently, you can set the :partition_by option. See "Ordering and partitioning".

• Rate limiting: Broadway allows developers to rate limit all producers in a single node by a given number of messages in a time period, allowing developers to easily work sources or sinks that cannot cope with a high number of requests. See the ":rate_limiting" option for producers in start_link/2.

• Metrics - Broadway uses the :telemetry library for instrumentation, see "Telemetry" section below for more information.

• Back-off (TODO)

The Broadway Behaviour

In order to use Broadway, you need to:

1. Define your pipeline configuration
2. Define a module implementing the Broadway behaviour

Example

Broadway is a process-based behaviour, and you begin by defining a module that invokes use Broadway. Processes defined by these modules will often be started by a supervisor, and so a start_link/1 function is frequently also defined but not strictly necessary.

defmodule MyBroadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(MyBroadway,
      name: MyBroadwayExample,
      producer: [
        module: {Counter, []},
        concurrency: 1
      ],
      processors: [
        default: [concurrency: 2]
      ]
    )
  end

  ...callbacks...
end

Then add your Broadway pipeline to your supervision tree (usually in lib/my_app/application.ex):

children = [
  {MyBroadway, []}
]

Supervisor.start_link(children, strategy: :one_for_one)

Adding your pipeline to your supervision tree in this way calls the default child_spec/1 function that is generated when use Broadway is invoked. If you would like to customize the child spec passed to the supervisor, you can override the child_spec/1 function in your module or explicitly pass a child spec to the supervisor when adding it to your supervision tree.

The configuration above defines a pipeline with:

• One producer
• Two processors

Here is how this pipeline would be represented:

                     [producer_1]
                         / \
                        /   \
                       /     \
                      /       \
             [processor_1] [processor_2]   <- process each message

After the pipeline is defined, you need to implement the handle_message/3 callback which will be invoked by processors for each message.

handle_message/3 receives every message as a Broadway.Message struct and it must return an updated message.

Batching

Depending on the scenario, you may want to group processed messages as batches before publishing your data. This is common and especially important when working with services like AWS S3 and SQS that provide a specific API for sending and retrieving batches. This can drastically increase throughput and consequently improve the overall performance of your pipeline.

To create batches, define the batchers configuration option:

defmodule MyBroadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(MyBroadway,
      name: MyBroadwayExample,
      producer: [
        module: {Counter, []},
        concurrency: 1
      ],
      processors: [
        default: [concurrency: 2]
      ],
      batchers: [
        sqs: [concurrency: 2, batch_size: 10],
        s3: [concurrency: 1, batch_size: 10]
      ]
    )
  end

  ...callbacks...
end

The configuration above defines a pipeline with:

• One producer
• Two processors
• One batcher named :sqs with two batch processors
• One batcher named :s3 with one batch processor

Here is how this pipeline would be represented:

                     [producer_1]
                         / \
                        /   \
                       /     \
                      /       \
             [processor_1] [processor_2]   <- process each message
                      /\     /\
                     /  \   /  \
                    /    \ /    \
                   /      x      \
                  /      / \      \
                 /      /   \      \
                /      /     \      \
           [batcher_sqs]    [batcher_s3]
                /\                  \
               /  \                  \
              /    \                  \
             /      \                  \
   [batch_sqs_1] [batch_sqs_2]    [batch_s3_1] <- process each batch

Additionally, define the handle_batch/4 callback, which batch processors invoke for each batch. You can then call Broadway.Message.put_batcher/2 inside handle_message/3 to control which batcher the message should go to.

The batcher receives processed messages and creates batches specified by the batch_size and batch_timeout configuration. The goal is to create a batch with at most batch_size entries within batch_timeout milliseconds. Each message goes into a particular batch, controlled by calling Broadway.Message.put_batch_key/2 in handle_message/3. Once a batch is created, it is sent to a separate process that will call handle_batch/4, passing the batcher, the batch itself (i.e. a list of messages), a Broadway.BatchInfo struct and the Broadway context.

For example, imagine your producer generates integers as data. You want to route the odd integers to SQS and the even ones to S3. Your pipeline would look like this:

defmodule MyBroadway do
  use Broadway
  import Integer

  alias Broadway.Message

  ...start_link...

  @impl true
  def handle_message(_, %Message{data: data} = message, _) when is_odd(data) do
    message
    |> Message.update_data(&process_data/1)
    |> Message.put_batcher(:sqs)
  end

  def handle_message(_, %Message{data: data} = message, _) when is_even(data) do
    message
    |> Message.update_data(&process_data/1)
    |> Message.put_batcher(:s3)
  end

  defp process_data(data) do
    # Do some calculations, generate a JSON representation, etc.
  end

  @impl true
  def handle_batch(:sqs, messages, _batch_info, _context) do
    # Send batch of successful messages as ACKs to SQS
    # This tells SQS that this list of messages were successfully processed
  end

  def handle_batch(:s3, messages, _batch_info, _context) do
    # Send batch of messages to S3
  end
end

See the callbacks documentation for more information on the arguments given to each callback and their expected return types.

Now you are ready to get started. See the start_link/2 function for a complete reference on the arguments and options allowed.

Also makes sure to check out GUIDES in the documentation sidebar for more examples, how tos and more.

Acknowledgements and failures

At the end of the pipeline, messages are automatically acknowledged.

If there are no batchers, the acknowledgement will be done by processors. The number of messages acknowledged, assuming the pipeline is running at full scale, will be max_demand - min_demand. Since the default values are 10 and 5 respectively, we will be acknowledging in groups of 5.

If there are batchers, the acknowledgement is done by the batchers, using the batch_size.

In case of failures, Broadway does its best to keep the failures contained and avoid losing messages. The failed message or batch is acknowledged as failed immediately. For every failure, a log report is also emitted. If your Broadway module also defines the handle_failed/2 callback, that callback will be invoked with all the failed messages before they get acknowledged.

Note however, that Broadway does not provide any sort of retries out of the box. This is left completely as a responsibility of the producer. For instance, if you are using Amazon SQS, the default behaviour is to retry unacknowledged messages after a user-defined timeout. If you don't want unacknowledged messages to be retried, is your responsibility to configure a dead-letter queue as target for those messages.

Testing

Many producers receive data from external systems and hitting the network is usually undesirable when running the tests.

For testing purposes, we recommend developers to use Broadway.DummyProducer. This producer does not produce any messages by itself and instead the test_message/3 and test_batch/3 functions should be used to publish messages.

With test_message/3, you can push a message into the pipeline and receive a process message when the pipeline acknowledges the data you have pushed has been processed.

Let's see an example. Imagine the following Broadway module:

defmodule MyBroadway do
  use Broadway

  def start_link() do
    producer_module = Application.fetch_env!(:my_app, :producer_module)

    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: producer_module
      ],
      processors: [
        default: []
      ],
      batchers: [
        default: [batch_size: 10]
      ]
    )
  end

  @impl true
  def handle_message(_processor, message, _context) do
    message
  end

  @impl true
  def handle_batch(_batcher, messages, _batch_info, _context) do
    messages
  end
end

Now in config/test.exs you could do:

config :my_app, :producer_module, {Broadway.DummyProducer, []}

And we can test it like this:

defmodule MyBroadwayTest do
  use ExUnit.Case, async: true

  test "test message" do
    ref = Broadway.test_message(MyBroadway, 1)
    assert_receive {:ack, ^ref, [%{data: 1}], []}
  end
end

Note that at the end we received a message in the format of:

{:ack, ^ref, successful_messages, failure_messages}

You can use the acknowledgment to guarantee the message has been processed and therefore any side-effect from the pipeline should be visible.

When using test_message/3, the message will be delivered as soon as possible, without waiting for the pipeline batch_size to be reached or without waiting for batch_timeout. This behaviour is useful to test and verify single messages, without imposing high timeouts to our test suites.

In case you want to test multiple messages, then you need to use test_batch/3. test_batch/3 will respect the batching configuration, which most likely means you need to increase your test timeouts:

test "batch messages" do
  {:ok, pid} = MyBroadway.start_link()
  ref = Broadway.test_batch(pid, [1, 2, 3])
  assert_receive {:ack, ^ref, [%{data: 1}, %{data: 2}, %{data: 3}], []}, 1000
end

However, keep in mind that, generally speaking, there is no guarantee the messages will arrive in the same order that you have sent them, especially for large batches, as Broadway will process large batches concurrently and order will be lost.

If you want to send more than one test message at once, then we recommend setting the :batch_mode to :bulk, especially if you want to assert how the code will behave with large batches. Otherwise the batcher will flush messages as soon as possible and in small batches.

However, keep in mind that, regardless of the :batch_mode you cannot rely on ordering, as Broadway pipelines are inherently concurrent. For example, if you send those messages:

test "multiple batch messages" do
  {:ok, pid} = MyBroadway.start_link()
  ref = Broadway.test_batch(pid, [1, 2, 3, 4, 5, 6, 7], batch_mode: :bulk)
  assert_receive {:ack, ^ref, [%{data: 1}], []}, 1000
end

Ordering and partitioning

By default, Broadway processes all messages and batches concurrently, which means ordering is not guaranteed. Some producers may impose some ordering (for instance, Apache Kafka), but if the ordering comes from a business requirement, you will have to impose the ordering yourself. This can be done with the :partition_by option, which enforces that messages with a given property are always forwarded to the same stage.

In order to provide partitioning throughout the whole pipeline, just set :partition_by at the root of your configuration:

defmodule MyBroadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(MyBroadway,
      name: MyBroadwayExample,
      producer: [
        module: {Counter, []},
        concurrency: 1
      ],
      processors: [
        default: [concurrency: 2]
      ],
      batchers: [
        sqs: [concurrency: 2, batch_size: 10],
        s3: [concurrency: 1, batch_size: 10]
      ],
      partition_by: &partition/1
    )
  end

  defp partition(msg) do
    msg.data.user_id
  end

In the example above, we are partitioning the pipeline by user_id. This means any message with the same user_id will be handled by the same processor and batch processor.

The partition function must return a non-negative integer, starting at zero, which is routed to a stage by using the remainder option.

If the data you want to partition by is not an integer, you can explicitly hash it by calling :erlang.phash2/1. However, note that hash does not guarantee an equal distribution of events across partitions. So some partitions may be more overloaded than others, slowing down the whole pipeline.

In the example above, we have set the same partition for all processors and batchers. You can also specify the :partition_by function for each "processor" and "batcher" individually.

Finally, beware of the error semantics when using partitioning. If you require ordering and a message fails, the partition will continue processing messages. Depending on the type of processing, the end result may be inconsistent. If your producer supports retrying, the failed message may be retried later, also out of order. Those issues happens regardless of Broadway and solutions to said problems almost always need to be addressed outside of Broadway too.

Telemetry

Broadway currently exposes following Telemetry events:

• [:broadway, :processor, :start] - Dispatched by a Broadway processor before the optional prepare_messages/2

  • Measurement: %{time: System.monotonic_time}
  • Metadata: %{name: atom, messages: [Broadway.Message.t]}

• [:broadway, :processor, :stop] - Dispatched by a Broadway processor after prepare_messages/2 and after all c:handle_message/2 callback has been invoked for all individual messages

  • Measurement: %{time: System.monotonic_time, duration: native_time}
  • Metadata:

    %{
      name: atom,
      successful_messages_to_ack: [Broadway.Message.t],
      successful_messages_to_forward: [Broadway.Message.t],
      failed_messages: [Broadway.Message.t]
    }

• [:broadway, :processor, :message, :start] - Dispatched by a Broadway processor before your handle_message/3 callback is invoked

  • Measurement: %{time: System.monotonic_time}
  • Metadata:

    %{
      processor_key: atom,
      name: atom,
      message: Broadway.Message.t
    }

• [:broadway, :processor, :message, :stop] - Dispatched by a Broadway processor after your handle_message/3 callback has returned

  • Measurement: %{time: System.monotonic_time, duration: native_time}
  • Metadata:

    %{
      processor_key: atom,
      name: atom,
      message: Broadway.Message.t,
      updated_message: Broadway.Message.t
    }

• [:broadway, :processor, :message, :exception] - Dispatched by a Broadway processor if your handle_message/3 callback encounters an exception

  • Measurement: %{time: System.monotonic_time, duration: native_time}
  • Metadata:

    %{
      processor_key: atom,
      name: atom,
      message: Broadway.Message.t,
      kind: kind,
      reason: reason,
      stacktrace: stacktrace
    }

• [:broadway, :consumer, :start] - Dispatched by a Broadway consumer before your handle_batch/4 callback is invoked

  • Measurement: %{time: System.monotonic_time}
  • Metadata:

    %{
      name: atom,
      messages: [Broadway.Message.t],
      batch_info: Broadway.BatchInfo.t
    }

• [:broadway, :consumer, :stop] - Dispatched by a Broadway consumer after your handle_batch/4 callback has returned

  • Measurement: %{time: System.monotonic_time, duration: native_time}
  • Metadata:

    %{
      name: atom,
      successful_messages: [Broadway.Message.t],
      failed_messages: [Broadway.Message.t],
      batch_info: Broadway.BatchInfo.t
    }

• [:broadway, :batcher, :start] - Dispatched by a Broadway batcher before handling events

  • Measurement: %{time: System.monotonic_time}
  • Metadata: %{name: atom, events: [{Broadway.Message.t}]}

• [:broadway, :batcher, :stop] - Dispatched by a Broadway batcher after handling events

  • Measurement: %{time: System.monotonic_time, duration: native_time}
  • Metadata: %{name: atom}