DOCSEARCHbyAlgolia
Submit Docs

brod

repository·master·Indexed 20 days ago

https://github.com/kafka4beam/brod

An Erlang implementation of the Apache Kafka protocol providing high-performance producers and consumers for Erlang and Elixir applications. It supports Apache Kafka v0.8+, featuring automatic batching, metadata refresh, and multiple consumer types including simple pollers, group subscribers, and topic subscribers. The library handles cluster disturbances like leader re-election internally and provides both synchronous and asynchronous production APIs.

Tokens
11K
Snippets
37
Records
40
Agent score
70%

What's inside brod

  1. Brod - Apache Kafka Client for Erlang/Elixir Overview

    master

    Brod is an Erlang implementation of the Apache Kafka protocol. It provides robust support for both producers and consumers, handling cluster disturbances like leader re-election internally.

    Key features include:

    • Supports Apache Kafka v0.8+.
    • Robust producer with automatic batching and automatic metadata refresh on errors (e.g., "Not a leader for partition").
    • Multiple consumer types: Simple poller (with configurable prefetch), Group subscriber (Kafka or custom offset storage), and Topic subscriber.
    • Direct APIs for message operations and cluster management without requiring full client/producer/consumer startup.
    • Configurable compression (no compression by default).
    • An escriptized CLI tool for management and inspection.

  2. How brod clients work

    master
    A brod_client is a gen_server responsible for maintaining TCP connections to Kafka brokers. It also manages producer and consumer processes for each topic-partition under a two-level supervision tree. To use any producer or consumer, you must first start at least one client to manage them.

  3. Handle producer acknowledgements (Acks)

    master

    For asynchronous produce APIs (brod:produce/3 and brod:produce/5), the caller receives a reply message for each call.

    To use the reply record, add -include_lib("brod/include/brod.hrl"). to your module.

    Reply Pattern

    #brod_produce_reply{ call_ref = CallRef, result = brod_produce_req_acked }

    Strategies for handling Acks

    1. Manual Tracking: If the caller is a gen_server, store CallRef in its state and match the reply against it when received.
    2. Blocking: Use brod:sync_produce_request(CallRef) to block the calling process until the ack is received.
    3. Callbacks: Use brod:produce_cb/4 or brod:produce_cb/6 to provide a callback function that executes when the ack is received.

    Important Notes:

    • If required_acks is set to none in the producer config, Kafka will not ack, and the reply is sent immediately after the message is sent to the socket.
    • Replies are only strictly ordered per-partition. If producing to multiple partitions, replies may arrive out of order relative to the call sequence.
    %% Using sync_produce_request to block for an ack
{ok, CallRef} = brod:produce(brod_client_1, <<"brod-test-topic-1">>, 0, <<"some-key">>, <<"some-value">>),
brod:sync_produce_request(CallRef).

  4. Use :hash for partition key routing

    master

    When calling :brod.produce_sync/5, providing :hash as the partition argument enables sticky routing based on the provided key.

    Internally, brod retrieves the partition count for the topic, generates a hash of the key using :erlang.phash2/1, and selects the partition using the remainder (rem) of the hash divided by the partition count. This mimics the behavior of Kafka's ProducerRecord where a key is present.

    # Manual implementation of what :hash does internally:
{:ok, count} = :brod.get_partitions_count(:kafka_client, topic)
partition = rem(:erlang.phash2(key), count)
:brod.produce_sync(:kafka_client, topic, partition, key, message)

# The preferred way using the :hash atom:
:brod.produce_sync(:kafka_client, topic, :hash, key, message)

  5. Implement a consumer using group_subscriber_v2

    master

    The group_subscriber_v2 implementation creates a worker for each partition of a topic, improving throughput and fault isolation.

    To use it, you must implement the :brod_group_subscriber_v2 behaviour.

    Required Callbacks:

    • init(arg, state): Initializes the subscriber state.
    • handle_message(message, state): Processes an incoming message. Returning {:ok, :commit, new_state} acknowledges the message to Kafka.

    Optional Callbacks:

    • assign_partitions(partitions, state)
    • get_committed_offset(partition, state)
    • terminate(reason, state)
    defmodule BrodSample.GroupSubscriberV2 do
  @behaviour :brod_group_subscriber_v2

  def init(_arg, _arg2) do
    {:ok, []}
  end

  def handle_message(message, state) do
    IO.inspect(message, label: "message")
    {:ok, :commit, []}
  end
end

  8. Configure and start a brod client

    master

    Before producing messages, you must start a client. You can configure clients globally in your application configuration (e.g., config/dev.exs) or start them dynamically.

    When configuring, you can specify :endpoints (a list of host/port tuples), :auto_start_producers (set to true to avoid manual producer management), and security settings like :ssl and :sasl.

    Note: :endpoints accepts multiple host/port tuples, such as [{"192.168.0.2", 9092}, {"192.168.0.3", 9092}].

    # Via configuration
import Config

config :brod,
  clients: [
    kafka_client: [
      endpoints: [localhost: 9092],
      auto_start_producers: true,
      ssl: true,
      sasl: {
        :plain,
        System.get_env("KAFKA_CLUSTER_API_KEY"),
        System.get_env("KAFKA_CLUSTER_API_SECRET")
      }
    ]
  ]

# OR via dynamic start
:brod.start_client([localhost: 9092], :kafka_client, auto_start_producers: true)

  9. Consume messages using Partition Subscribers

    master

    Kafka consumers in Brod work in poll mode via a brod_consumer (the poller). A Partition Subscriber provides the highest level of flexibility by working directly with per-partition pollers.

    Messages are delivered to subscribers in message sets (batches), not individual messages. However, subscribers are permitted to acknowledge individual offsets.

    Implementation Steps

    1. Start the client: brod:start_client/2.
    2. Start a consumer: brod:start_consumer/3.
    3. Subscribe to a specific partition: brod:subscribe/5.
    ok = brod:start_client([{"localhost", 9092}], my_client),
ok = brod:start_consumer(my_client, <<"my_topic">>, []).

%% In a separate process for each partition:
{ok, ConsumerPid} = brod:subscribe(my_client, self(), <<"my_topic">>, Partition, []).

  10. Build and test Brod

    master

    To build and test Brod, ensure you have the following prerequisites:

    • Erlang/OTP version 24 or higher.
    • CMake 4 (required for building the crc32cer NIF).
    • docker-compose (required for running make test-env t).

    Use the following commands:

    make compile
make test-env t

  11. Consume messages using Partition Subscriber

    master

    For fine-grained control or when consuming from a single partition, use the low-level partition subscription approach.

    Workflow:

    1. Start a consumer for the topic using :brod.start_consumer/3.
    2. Subscribe to the specific partition using :brod.subscribe/5. This returns a consumer_pid.
    3. Handle incoming messages in a GenServer via handle_info/2. Messages arrive as a kafka_message_set.
    4. Acknowledge processed messages using :brod.consume_ack(consumer_pid, offset).
    5. Handle errors (like kafka_fetch_error) in handle_info/2 to manage consumer lifecycle or crashes.
    defmodule BrodSample.PartitionSubscriber do
  use GenServer

  import Record, only: [defrecord: 2, extract: 2]

  defrecord :kafka_message, extract(:kafka_message, from_lib: "brod/include/brod.hrl")
  defrecord :kafka_message_set, extract(:kafka_message_set, from_lib: "brod/include/brod.hrl")
  defrecord :kafka_fetch_error, extract(:kafka_fetch_error, from_lib: "brod/include/brod.hrl")

  defmodule State do
    @enforce_keys [:consumer_pid]
    defstruct consumer_pid: nil
  end

  defmodule KafkaMessage do
    @enforce_keys [:offset, :key, :value, :ts]
    defstruct offset: nil, key: nil, value: nil, ts: nil
  end

  def start_link(topic, partition) do
    GenServer.start_link(__MODULE__, {topic, partition})
  end

  @impl true
  def init({topic, partition}) do
    :ok = :brod.start_consumer(:kafka_client, topic, begin_offset: :latest)
    {:ok, consumer_pid} = :brod.subscribe(:kafka_client, self(), topic, partition, [])
    {:ok, %State{consumer_pid: consumer_pid}}
  end

  @impl true
  def handle_info(
        {consumer_pid, kafka_message_set(messages: msgs)},
        %State{consumer_pid: consumer_pid} = state
      ) do
    for msg <- msgs do
      msg = kafka_message_to_struct(msg)
      IO.inspect(msg)
      :brod.consume_ack(consumer_pid, msg.offset)
    end

    {:noreply, state}
  end

  def handle_info({pid, kafka_fetch_error()} = error, %State{consumer_pid: pid} = state) do
    {:stop, error, state}
  end

  defp kafka_message_to_struct(kafka_message(offset: offset, key: key, value: value, ts: ts)) do
    %KafkaMessage{
      offset: offset,
      key: key,
      value: value,
      ts: DateTime.from_unix!(ts, :millisecond)
    }
  end
end

  12. Enable Compression in Brod

    master

    Brod does not include a compression/decompression implementation by default. To enable compression, you must add a compression library (e.g., snappyer) as a dependency in your project's rebar.config.

    {deps, [
    {snappyer, "1.2.9"}
]}.