How to Monitor Brod with Vigilmon

Brod is a battle-tested, high-performance Erlang/Elixir Apache Kafka client used at scale in production. It supports producer groups, consumer groups, offset management, transaction semantics, and handles the full breadth of Kafka protocol features. Unlike higher-level wrappers, Brod gives you direct control over partitions, offsets, and group coordination — which means production failures are also more direct: a misconfigured consumer group stops processing silently, a producer timeout leaves messages unacknowledged, or a broker rebalance stalls your pipeline while BEAM supervisors report everything as green.

Vigilmon adds the external layer your Kafka client needs: heartbeat monitors that fire when consumers are processing, HTTP monitors for broker health APIs, and Slack alerts when consumer lag grows or producers fall behind.

This tutorial covers:

• Heartbeat monitoring from Brod consumer group callbacks
• HTTP monitoring for Kafka broker health endpoints
• Consumer lag alerting via heartbeat suppression
• Producer delivery tracking and alerting
• Status pages for Kafka-dependent services

What to Monitor in a Brod Kafka Deployment

| Layer | What it is | How to monitor | |---|---|---| | Consumer group | Brod consumer callbacks processing messages | Heartbeat per batch processed | | Producer delivery | Messages acknowledged by the broker | Heartbeat on successful produce | | Broker health | Kafka broker HTTP/JMX API | HTTP uptime check | | Consumer lag | Offset commit progress vs. partition head | Heartbeat suppressed when lag grows | | Group coordinator | Consumer group rebalance events | Heartbeat on stable rebalance |

Step 1: Add heartbeat monitoring to a Brod consumer

Brod consumers implement a callback module. Add a Vigilmon heartbeat ping inside the message handler so the heartbeat is only sent when real messages are being processed:

# lib/my_app/kafka/order_consumer.ex
defmodule MyApp.Kafka.OrderConsumer do
  @behaviour :brod_group_subscriber_v2

  require Logger

  # Brod group subscriber v2 init callback
  def init(_group_id, _member_id, _topic_partitions, _config) do
    {:ok, %{processed_count: 0, last_heartbeat_at: nil}}
  end

  # Called for each message set from a partition
  def handle_message(
        _topic,
        _partition,
        %{:offset => offset, :value => value} = msg,
        state
      ) do
    case process_order(value) do
      {:ok, _result} ->
        count = state.processed_count + 1
        state = %{state | processed_count: count}

        # Ping heartbeat every 100 messages or every 30 seconds
        state = maybe_ping_heartbeat(state)

        # Acknowledge the offset
        {:ok, :ack, state}

      {:error, reason} ->
        Logger.error("Failed to process order at offset #{offset}: #{inspect(reason)}")
        # Nack without retry to avoid stalling the partition
        {:ok, :ack_no_commit, state}
    end
  end

  defp maybe_ping_heartbeat(%{processed_count: count, last_heartbeat_at: last} = state) do
    now = System.monotonic_time(:second)
    due_by_count = rem(count, 100) == 0
    due_by_time = is_nil(last) or now - last > 30

    if due_by_count or due_by_time do
      ping_vigilmon()
      %{state | last_heartbeat_at: now}
    else
      state
    end
  end

  defp ping_vigilmon do
    url = Application.get_env(:my_app, :vigilmon_consumer_heartbeat_url)

    if url do
      case :httpc.request(:get, {String.to_charlist(url), []}, [{:timeout, 5000}], []) do
        {:ok, _} -> :ok
        {:error, reason} -> Logger.warning("Vigilmon heartbeat failed: #{inspect(reason)}")
      end
    end
  end

  defp process_order(value) do
    # Decode and process the Kafka message value
    case Jason.decode(value) do
      {:ok, order} -> MyApp.Orders.process(order)
      {:error, _} -> {:error, :invalid_json}
    end
  end
end

Start the consumer group in your supervision tree:

# lib/my_app/application.ex
def start(_type, _args) do
  children = [
    # ... your other children
    {MyApp.Kafka.Supervisor, []}
  ]
  Supervisor.start_link(children, strategy: :one_for_one)
end

# lib/my_app/kafka/supervisor.ex
defmodule MyApp.Kafka.Supervisor do
  use Supervisor

  def start_link(_opts), do: Supervisor.start_link(__MODULE__, [], name: __MODULE__)

  @impl true
  def init(_init_arg) do
    brokers = Application.get_env(:my_app, :kafka_brokers)
    group_id = "my_app_orders"
    topics = ["orders"]

    consumer_config = [
      begin_offset: :earliest,
      offset_commit_policy: :commit_to_kafka_v2,
      offset_commit_interval_seconds: 10
    ]

    group_config = [
      offset_commit_policy: :commit_to_kafka_v2,
      offset_commit_interval_seconds: 10
    ]

    children = [
      %{
        id: :brod_client,
        start: {:brod, :start_link_client, [brokers, :kafka_client, []]}
      },
      %{
        id: :order_consumer,
        start: {
          :brod_group_subscriber_v2,
          :start_link,
          [
            %{
              client: :kafka_client,
              group_id: group_id,
              topics: topics,
              cb_module: MyApp.Kafka.OrderConsumer,
              init_data: [],
              message_type: :message_set,
              consumer_config: consumer_config,
              group_config: group_config
            }
          ]
        }
      }
    ]

    Supervisor.init(children, strategy: :one_for_one)
  end
end

Configure your brokers and heartbeat URL:

# config/runtime.exs
config :my_app,
  kafka_brokers: [{"kafka.internal", 9092}],
  vigilmon_consumer_heartbeat_url: System.get_env("VIGILMON_CONSUMER_HEARTBEAT_URL"),
  vigilmon_producer_heartbeat_url: System.get_env("VIGILMON_PRODUCER_HEARTBEAT_URL")

In Vigilmon:

1. Click New Monitor → Heartbeat.
2. Name it Brod order consumer.
3. Set interval to 2 minutes (covers burst gaps in low-traffic periods).
4. Copy the heartbeat URL into your deployment config.

Step 2: Track producer delivery with a heartbeat

For critical Kafka producers, send a heartbeat after each successful produce acknowledgement:

# lib/my_app/kafka/event_producer.ex
defmodule MyApp.Kafka.EventProducer do
  require Logger

  @topic "events"

  def produce(event) when is_map(event) do
    key = Map.get(event, :id, "") |> to_string()
    value = Jason.encode!(event)

    case :brod.produce_sync(:kafka_client, @topic, :hash, key, value) do
      :ok ->
        ping_producer_heartbeat()
        {:ok, event}

      {:error, reason} ->
        Logger.error("Kafka produce failed for topic #{@topic}: #{inspect(reason)}")
        {:error, reason}
    end
  end

  defp ping_producer_heartbeat do
    url = Application.get_env(:my_app, :vigilmon_producer_heartbeat_url)

    if url do
      # Fire-and-forget: don't block the produce path on the heartbeat
      Task.start(fn ->
        :httpc.request(:get, {String.to_charlist(url), []}, [{:timeout, 5000}], [])
      end)
    end
  end
end

For high-throughput producers, throttle the heartbeat to once per N messages rather than per-message:

# Use an ETS counter or Agent to rate-limit pings
defp ping_producer_heartbeat do
  count = :ets.update_counter(:producer_counter, :count, {2, 1, 999, 0}, {1, 0})
  if count == 0, do: do_ping()
end

Step 3: Monitor Kafka broker health

Kafka brokers expose a health endpoint. Monitor it with Vigilmon's HTTP monitor:

For Confluent Cloud:

1. Click New Monitor → HTTP.
2. URL: https://api.confluent.cloud/v2/clusters (or your cluster REST endpoint).
3. Check interval: 60 seconds.

For self-hosted Kafka with the REST Proxy:

# Install kafka-rest if not already running
# Then monitor:
http://your-kafka-rest:8082/brokers

1. Click New Monitor → HTTP.
2. URL: http://your-kafka-rest.internal:8082/brokers.
3. Expected status: 200.
4. Add a keyword check for the broker ID (e.g., "0") to verify the response is valid.

For Kafka with Zookeeper (older deployments), monitor the Zookeeper health endpoint:

# Zookeeper 4-letter word — ruok
echo ruok | nc zookeeper.internal 2181
# Responds: imok

Wrap this in a simple script and expose it via a lightweight HTTP endpoint your team controls.

Step 4: Detect consumer lag via heartbeat suppression

Consumer lag is difficult to alert on directly without a Kafka metrics exporter. Vigilmon's heartbeat model gives you a simple proxy: if the consumer is processing messages, heartbeats arrive. If lag grows past a threshold (consumer falls behind), processing slows and pings thin out.

Make this explicit by checking lag before pinging:

defp maybe_ping_heartbeat(state) do
  case get_consumer_lag() do
    {:ok, lag} when lag < 10_000 ->
      ping_vigilmon()

    {:ok, lag} ->
      Logger.warning("Consumer lag #{lag} exceeds threshold — suppressing heartbeat")
      # No ping → Vigilmon alerts when grace period expires

    {:error, _reason} ->
      # Lag check failed — still ping so we don't false-alarm on metric fetch issues
      ping_vigilmon()
  end

  state
end

defp get_consumer_lag do
  # Fetch your partition's current offset and the high watermark
  # Using brod's offset APIs or an external lag monitor
  with {:ok, committed} <- :brod.fetch_committed_offsets(:kafka_client, "my_app_orders"),
       {:ok, latest} <- get_partition_latest_offset() do
    lag = latest - committed
    {:ok, lag}
  end
end

Step 5: Configure alerts

In Vigilmon go to Alerts and configure notification channels:

• Slack (#kafka-ops or #data-engineering) — consumer and producer failures
• PagerDuty — for pipelines where message loss is business-critical
• Email — backend engineering team

Recommended alert policies for Brod:

| Monitor | Condition | Action | |---|---|---| | Consumer heartbeat | Missed ≥ 2× | Alert data engineering | | Producer heartbeat | Missed ≥ 2× | Alert backend on-call | | Kafka broker HTTP | Non-200 or timeout | Alert infrastructure team | | Consumer lag signal | Heartbeat suppressed | Warning — lag growing | | Group rebalance | Heartbeat gap > 5 min | Investigate rebalance storm |

Step 6: Status page for Kafka-dependent services

Create a Vigilmon status page that downstream teams can check:

1. Go to Status Pages → New Page.
2. Add: Kafka broker HTTP monitor, consumer heartbeat monitor, producer heartbeat monitor.
3. Set visibility to Team (internal) or Public if downstream partners depend on your pipeline.
4. Add the status page URL to your service runbook.

Key Metrics to Watch

| Metric | Threshold | Meaning | |---|---|---| | Consumer heartbeat | < 2 missed | Consumer processing messages | | Producer heartbeat | < 2 missed | Producer delivering to broker | | Broker HTTP status | Must be 200 | Kafka cluster reachable | | Consumer lag proxy | Heartbeat not suppressed | Lag under threshold | | Offset commit latency | < 10 s | No coordinator issues | | Rebalance events | < 3/hour | Consumer group stable |

Why Monitor Brod?

Kafka client failures are operationally silent:

• Consumer group stall — a rebalance that never completes leaves partitions unassigned; supervisors report the processes as healthy
• Producer timeout — produce_sync returning {:error, :timeout} means messages never reach the broker; no BEAM alarm fires
• Offset commit failure — consumers process messages but fail to commit; on restart, the same messages are reprocessed indefinitely
• Broker leadership change — a partition leader election causes a temporary producer gap that looks like normal operation but loses messages

Vigilmon's heartbeat pattern makes these visible: the consumer only pings when it completes successful message processing. Any failure in the pipeline — group stall, broker timeout, offset commit error — stops the heartbeats and triggers your alert.

Conclusion

Brod is the production-grade Erlang/Elixir Kafka client for teams that need direct control over Kafka protocol semantics. Vigilmon is the external monitoring layer that tells you when Brod's consumers, producers, and brokers are actually working — not just alive. With consumer heartbeats, producer delivery tracking, broker HTTP monitoring, and consumer lag signals, you have the observability depth that high-throughput Kafka deployments require.

Get started free at vigilmon.online.