How to Monitor KafkaEx with Vigilmon

KafkaEx is the Elixir client for Apache Kafka, providing producer, consumer, metadata fetching, offset management, and consumer group coordination for stream processing applications. It integrates naturally with OTP supervision trees and makes Kafka feel idiomatic in Elixir.

Kafka is built for durability — messages are persisted, partitions replicated, consumer group offsets committed. But the connection between your Elixir app and the Kafka cluster is not automatically observable: consumer lag builds silently, partition leadership changes can pause consumption, and network splits can cause your producers to buffer indefinitely.

This tutorial adds production observability to a KafkaEx application:

• A health check endpoint that verifies Kafka connectivity and consumer group health
• HTTP uptime monitoring with Vigilmon
• Heartbeat monitoring for consumer workers and producer pipelines
• Slack alerts for broker connectivity and consumer lag events
• A status page for your streaming infrastructure

Why Monitor KafkaEx Applications?

KafkaEx provides resilient reconnection, but monitoring gaps are common:

• Consumer lag buildup: consumers are running and healthy by OTP's definition, but they've fallen behind producers. Messages pile up in partitions. No alert fires until application-level side effects appear.
• Partition rebalance storm: consumer group rebalancing during a rolling deploy causes all consumers to briefly pause. If the broker is slow to acknowledge rebalance completion, consumption stops for minutes.
• Broker metadata staleness: your KafkaEx worker has cached stale partition metadata and is writing to a partition whose leader changed. Produce calls fail until metadata is refreshed.
• Dead-letter queue buildup: messages that fail deserialization or processing are routed to a dead-letter topic. Without monitoring, this queue grows indefinitely.

External monitoring — health checks for connectivity and heartbeats for end-to-end processing — catches all of these.

Step 1: Add a Health Check Endpoint

The health check should verify that your app can reach Kafka and that your consumer workers are running.

# lib/my_app_web/plugs/health_check.ex
defmodule MyAppWeb.Plugs.HealthCheck do
  import Plug.Conn
  require Logger

  def init(opts), do: opts

  def call(%Plug.Conn{request_path: "/health"} = conn, _opts) do
    checks = %{
      database: check_database(),
      kafka: check_kafka_connectivity(),
      consumers: check_consumer_workers()
    }

    status = if Enum.all?(checks, fn {_, v} -> v == :ok end), do: 200, else: 503

    conn
    |> put_resp_content_type("application/json")
    |> send_resp(status, Jason.encode!(%{
      status: if(status == 200, do: "ok", else: "degraded"),
      checks: checks
    }))
    |> halt()
  end

  def call(conn, _opts), do: conn

  defp check_database do
    case Ecto.Adapters.SQL.query(MyApp.Repo, "SELECT 1", []) do
      {:ok, _} -> :ok
      _ -> :error
    end
  end

  defp check_kafka_connectivity do
    # Fetch metadata as a connectivity probe — lightweight and accurate
    case KafkaEx.metadata(topic: "health-check-probe") do
      %KafkaEx.Protocol.Metadata.Response{} -> :ok
      _ -> :error
    end
  rescue
    _ -> :error
  catch
    :exit, _ -> :error
  end

  defp check_consumer_workers do
    workers = [
      MyApp.Consumers.EventConsumer,
      MyApp.Consumers.AuditConsumer
    ]

    all_alive = Enum.all?(workers, fn mod ->
      case Process.whereis(mod) do
        nil -> false
        pid -> Process.alive?(pid)
      end
    end)

    if all_alive, do: :ok, else: :error
  end
end

Add it to your Phoenix endpoint before the router:

# lib/my_app_web/endpoint.ex
defmodule MyAppWeb.Endpoint do
  use Phoenix.Endpoint, otp_app: :my_app

  plug MyAppWeb.Plugs.HealthCheck

  plug MyAppWeb.Router
end

Test it:

mix phx.server
curl http://localhost:4000/health
# {"status":"ok","checks":{"database":"ok","kafka":"ok","consumers":"ok"}}

Step 2: Set Up HTTP Monitoring in Vigilmon

1. Sign up at vigilmon.online
2. Click New Monitor → HTTP
3. Enter https://yourdomain.com/health
4. Set the check interval to 1 minute
5. Optionally add a keyword check that the body contains "ok" to detect degraded responses
6. Save

Multi-region monitoring from Vigilmon is valuable for Kafka apps: if your broker cluster is in a specific region, Vigilmon will catch network path issues between regions that your in-cluster health checks wouldn't see.

Step 3: Heartbeat Monitoring for Consumers

Consumer heartbeats are the most important signal in a Kafka application. A consumer that is alive but has zero throughput is worse than a consumer that crashed — at least a crash triggers a restart.

Consumer Worker with Heartbeat

# lib/my_app/consumers/event_consumer.ex
defmodule MyApp.Consumers.EventConsumer do
  use GenServer
  require Logger

  @topic "app.events"
  @consumer_group "event-processor"
  @poll_interval 1_000
  @heartbeat_every_n_messages 50

  def start_link(_), do: GenServer.start_link(__MODULE__, [], name: __MODULE__)

  @impl true
  def init(_) do
    worker = start_kafka_worker()
    Process.send_after(self(), :poll, @poll_interval)
    {:ok, %{worker: worker, processed: 0, since_last_heartbeat: 0}}
  end

  @impl true
  def handle_info(:poll, %{worker: worker} = state) do
    messages = KafkaEx.fetch(@topic,
      0,  # partition
      worker_name: worker,
      auto_commit: false
    )

    new_state = Enum.reduce(messages, state, fn
      %KafkaEx.Protocol.Fetch.Response{partitions: partitions}, acc ->
        Enum.reduce(partitions, acc, fn partition, inner_acc ->
          Enum.reduce(partition.message_set, inner_acc, fn msg, s ->
            case process_message(msg.value) do
              :ok ->
                KafkaEx.offset_commit(worker, %KafkaEx.Protocol.OffsetCommit.Request{
                  consumer_group: @consumer_group,
                  topic: @topic,
                  partition: partition.partition,
                  offset: msg.offset + 1
                })

                count = s.since_last_heartbeat + 1
                if count >= @heartbeat_every_n_messages do
                  ping_heartbeat()
                  %{s | processed: s.processed + 1, since_last_heartbeat: 0}
                else
                  %{s | processed: s.processed + 1, since_last_heartbeat: count}
                end

              {:error, reason} ->
                Logger.error("Failed to process message at offset #{msg.offset}: #{inspect(reason)}")
                route_to_dead_letter(msg)
                s
            end
          end)
        end)
      _, acc -> acc
    end)

    Process.send_after(self(), :poll, @poll_interval)
    {:noreply, new_state}
  end

  defp start_kafka_worker do
    KafkaEx.create_worker(
      :event_consumer_worker,
      uris: Application.fetch_env!(:my_app, :kafka_brokers),
      consumer_group: @consumer_group,
      consumer_group_update_interval: 100
    )
    :event_consumer_worker
  end

  defp process_message(payload) do
    case Jason.decode(payload) do
      {:ok, event} -> MyApp.Events.handle(event)
      {:error, _} -> {:error, :invalid_json}
    end
  end

  defp route_to_dead_letter(msg) do
    KafkaEx.produce(%KafkaEx.Protocol.Produce.Request{
      topic: "app.events.dead-letter",
      partition: 0,
      required_acks: 1,
      messages: [%KafkaEx.Protocol.Produce.Message{value: msg.value}]
    })
  end

  defp ping_heartbeat do
    url = Application.get_env(:my_app, :vigilmon)[:event_consumer_heartbeat_url]
    if url do
      case Req.get(url, receive_timeout: 5_000) do
        {:ok, _} -> :ok
        {:error, reason} -> Logger.warning("Heartbeat ping failed: #{inspect(reason)}")
      end
    end
  end
end

For low-throughput topics, use a timer-based heartbeat instead:

# In init/1, schedule periodic heartbeat
Process.send_after(self(), :heartbeat_tick, :timer.minutes(5))

# Handle it
def handle_info(:heartbeat_tick, state) do
  if state.processed_since_last_tick > 0 do
    ping_heartbeat()
    Logger.info("Heartbeat: processed #{state.processed_since_last_tick} messages")
  else
    Logger.warning("Heartbeat: no messages processed in the last 5 minutes")
    # Still ping to signal the consumer is alive but idle (adjust based on expected traffic)
  end

  Process.send_after(self(), :heartbeat_tick, :timer.minutes(5))
  {:noreply, %{state | processed_since_last_tick: 0}}
end

Set the heartbeat URL:

# config/runtime.exs
config :my_app, :vigilmon,
  event_consumer_heartbeat_url: System.get_env("VIGILMON_EVENT_CONSUMER_HEARTBEAT_URL")

Create the Heartbeat Monitor

1. In Vigilmon, click New Monitor → Heartbeat
2. Set expected interval based on your traffic: for @heartbeat_every_n_messages 50 and ~100 msg/min throughput, set to 1 minute
3. Copy the ping URL and set it as VIGILMON_EVENT_CONSUMER_HEARTBEAT_URL
4. Save

Step 4: Producer Heartbeat

Monitor your Kafka producers the same way — ping on each successful produce batch:

# lib/my_app/publishers/event_publisher.ex
defmodule MyApp.Publishers.EventPublisher do
  use GenServer
  require Logger

  @topic "app.events"
  @flush_interval :timer.seconds(5)

  def start_link(_), do: GenServer.start_link(__MODULE__, [], name: __MODULE__)
  def publish(event), do: GenServer.cast(__MODULE__, {:publish, event})

  @impl true
  def init(_) do
    Process.send_after(self(), :flush, @flush_interval)
    {:ok, %{buffer: []}}
  end

  @impl true
  def handle_cast({:publish, event}, state) do
    {:noreply, %{state | buffer: [event | state.buffer]}}
  end

  def handle_info(:flush, %{buffer: []} = state) do
    Process.send_after(self(), :flush, @flush_interval)
    {:noreply, state}
  end

  def handle_info(:flush, %{buffer: buffer} = state) do
    messages = Enum.map(buffer, fn event ->
      %KafkaEx.Protocol.Produce.Message{value: Jason.encode!(event)}
    end)

    case KafkaEx.produce(%KafkaEx.Protocol.Produce.Request{
      topic: @topic,
      partition: 0,
      required_acks: 1,
      messages: messages
    }) do
      :ok ->
        ping_heartbeat()
        Logger.debug("Published #{length(messages)} events to #{@topic}")
      {:error, reason} ->
        Logger.error("Failed to publish to #{@topic}: #{inspect(reason)}")
        # Re-buffer on failure (implement retry logic appropriate for your use case)
    end

    Process.send_after(self(), :flush, @flush_interval)
    {:noreply, %{state | buffer: []}}
  end

  defp ping_heartbeat do
    url = Application.get_env(:my_app, :vigilmon)[:event_publisher_heartbeat_url]
    if url, do: Req.get(url, receive_timeout: 5_000)
  end
end

Step 5: Alerts via Slack

In Vigilmon, go to Notifications → New Channel → Slack and paste your Slack webhook URL.

Assign it to all monitors. When the Kafka health check fails:

🔴 DOWN: yourdomain.com/health
Checks degraded: kafka
Detected from: EU-West, US-East
2 minutes ago

When a consumer heartbeat is missed:

🔴 HEARTBEAT MISSED: event-consumer
Last ping: 12 minutes ago (expected: every 1 minute)

The heartbeat miss alert is often more actionable than a health check failure because it tells you exactly which consumer stalled.

Step 6: Status Page

1. Go to Status Pages → New Status Page in Vigilmon
2. Add monitors:
   • "API Health" — HTTP health endpoint
   • "Event Consumer" — consumer heartbeat
   • "Event Publisher" — producer heartbeat
3. Copy the public URL and share with your team

Key Metrics for KafkaEx Applications

| Monitor | What it catches | |---------|----------------| | Health endpoint Kafka check | Broker metadata fetch works | | Consumer heartbeat (per topic/partition) | Actual message processing throughput | | Producer heartbeat | Events are being produced successfully | | Response time on health check | Slow metadata fetch = broker pressure | | Heartbeat cadence change | Throughput drop before complete stall |

Dead-Letter Queue Monitoring

If your consumers route failed messages to a dead-letter topic, add a heartbeat that fires when the dead-letter queue is empty or below threshold:

defp maybe_ping_dlq_health do
  # Ping a separate "DLQ health" heartbeat only if DLQ has been empty this cycle
  dlq_count = count_dead_letter_messages()
  if dlq_count == 0 do
    url = Application.get_env(:my_app, :vigilmon)[:dlq_health_heartbeat_url]
    if url, do: Req.get(url, receive_timeout: 5_000)
  end
end

A missed heartbeat on the DLQ health monitor means messages are failing — even if the consumer itself is running fine.

What You've Built

| What | How | |------|-----| | Kafka connectivity check | KafkaEx metadata fetch as health probe | | Consumer process check | All consumer GenServers alive | | HTTP uptime monitoring | Vigilmon HTTP monitor → /health | | Consumer heartbeat | Ping every N messages or on timer | | Producer heartbeat | Ping after each successful flush | | Dead-letter queue monitoring | Heartbeat on DLQ empty cycles | | Slack alerts | Per-monitor down + heartbeat miss notifications | | Status page | Unified streaming infrastructure view |

Kafka retains your messages. Vigilmon tells you when your consumers have stopped reading them.

Next Steps

• Add one heartbeat monitor per Kafka topic partition group for fine-grained lag visibility
• Use Vigilmon's response time history to detect metadata fetch slowdowns that precede broker issues
• Add a separate monitor for your Kafka broker management UI or health endpoint if you use Confluent Control Center or Redpanda Console

Get started free at vigilmon.online.