tutorial

How to Monitor KafkaEx with Vigilmon

Add uptime monitoring, heartbeat checks, and alerts to your Elixir Kafka application using KafkaEx — covering broker connectivity, consumer group lag, and streaming message pipelines.

How to Monitor KafkaEx with Vigilmon

KafkaEx is an Apache Kafka client library for Elixir. It provides consumer groups, producer compression, offset management, and streaming message processing with an Elixir-idiomatic API — a common alternative to Brod for teams that prefer working with Elixir conventions over Erlang ones.

Kafka-based applications fail quietly. Consumer group lag builds up silently, broker connections drop and recover without alerting anything, and workers that process messages can crash without producing user-visible errors. External monitoring is the only way to catch these failures before they cause data loss or delayed processing.

This tutorial adds production observability to a KafkaEx application:

  • A health check HTTP endpoint that verifies broker connectivity
  • HTTP uptime monitoring with Vigilmon
  • Heartbeat monitoring for consumer workers and streaming pipelines
  • Slack alerts and a status page

Why Monitor a KafkaEx Application?

Kafka consumers fail in ways that don't surface as obvious errors:

  • Consumer group rebalance loop: consumers join and leave rapidly, preventing any partition from being processed
  • Offset commit failure: messages are consumed but offsets aren't committed, causing duplicate processing after restart
  • Broker partition leadership change: the consumer connects to a non-leader broker and receives no messages
  • GenStage pipeline backpressure: upstream stages stall downstream ones, messages stop flowing without errors
  • Worker process crash: the KafkaEx consumer supervisor restarts the worker, but messages delivered during that window may be re-delivered or lost depending on offset commit timing

Vigilmon catches all of these by combining HTTP health checks (is the app running?) with heartbeat monitoring (is work actually happening?).

Key Metrics to Monitor

| Metric | Why It Matters | |--------|---------------| | Broker connectivity | Can the app reach the Kafka cluster? | | Consumer group lag | Are consumers keeping up with producers? | | Worker process liveness | Are consumer workers running? | | Message throughput | Is the pipeline processing messages? | | Offset commit latency | Are offsets being committed promptly? |

Step 1: Add a health check endpoint

Add a health check endpoint that Vigilmon can poll to verify the application is operational.

# mix.exs
defp deps do
  [
    {:kafka_ex, "~> 0.13"},
    {:bandit, "~> 1.0"},
    {:jason, "~> 1.4"},
    {:req, "~> 0.5"}
  ]
end

Create a health check plug that tests broker connectivity:

# lib/my_app/health_check.ex
defmodule MyApp.HealthCheck do
  import Plug.Conn

  def init(opts), do: opts

  def call(%Plug.Conn{request_path: "/health"} = conn, _opts) do
    checks = %{
      kafka_broker: check_kafka_broker(),
      consumer_workers: 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: status_label(status), checks: checks}))
    |> halt()
  end

  def call(conn, _opts), do: conn

  defp check_kafka_broker do
    # Attempt to list topics as a liveness probe
    case KafkaEx.metadata(topic: "health-check") do
      %KafkaEx.Protocol.Metadata.Response{} -> :ok
      _ -> :error
    end
  rescue
    _ -> :error
  catch
    :exit, _ -> :error
  end

  defp check_consumer_workers do
    # Verify at least one consumer worker is alive
    case Process.whereis(MyApp.ConsumerSupervisor) do
      pid when is_pid(pid) ->
        children = Supervisor.which_children(pid)
        if Enum.any?(children, fn {_, child_pid, _, _} ->
          is_pid(child_pid) and Process.alive?(child_pid)
        end), do: :ok, else: :error
      nil -> :error
    end
  end

  defp status_label(200), do: "ok"
  defp status_label(_), do: "degraded"
end

Wire it up:

# lib/my_app/http_server.ex
defmodule MyApp.HttpServer do
  use Plug.Router

  plug MyApp.HealthCheck
  plug :match
  plug :dispatch

  match _ do
    send_resp(conn, 404, "not found")
  end
end

Add to your supervision tree:

# lib/my_app/application.ex
def start(_type, _args) do
  children = [
    KafkaEx.Supervisor,
    MyApp.ConsumerSupervisor,
    {Bandit, plug: MyApp.HttpServer, port: 4000}
  ]

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

Test it:

curl http://localhost:4000/health
# {"status":"ok","checks":{"kafka_broker":"ok","consumer_workers":"ok"}}

Step 2: Set up HTTP monitoring in Vigilmon

Point Vigilmon at your health endpoint:

  1. Sign up at vigilmon.online
  2. Click New Monitor → HTTP
  3. Enter https://yourdomain.com/health
  4. Set check interval: 1 minute (paid) or 5 minutes (free)
  5. Add a keyword check: "kafka_broker":"ok" must be present
  6. Save

The keyword check distinguishes between "HTTP server is up but Kafka is unreachable" and "fully healthy." Without it, a 200 response with a degraded Kafka connection would look healthy.

Step 3: Heartbeat monitoring for consumer workers

Consumer group lag is the most dangerous failure mode for Kafka applications. If consumers stop processing messages, the lag grows silently until it becomes a data loss event or causes downstream timeouts.

KafkaEx consumer heartbeat

# lib/my_app/consumer.ex
defmodule MyApp.Consumer do
  use KafkaEx.GenConsumer

  require Logger

  def init(topic, partition) do
    {:ok, %{topic: topic, partition: partition, last_processed: System.monotonic_time()}}
  end

  def handle_message_set(message_set, state) do
    Enum.each(message_set, fn message ->
      case process_message(message) do
        :ok ->
          :ok
        {:error, reason} ->
          Logger.error("Failed to process message offset=#{message.offset}: #{inspect(reason)}")
      end
    end)

    ping_heartbeat()
    {:async_commit, %{state | last_processed: System.monotonic_time()}}
  end

  defp process_message(_message), do: :ok  # your logic

  defp ping_heartbeat do
    url = Application.get_env(:my_app, :vigilmon)[:consumer_heartbeat_url]
    if url, do: Task.start(fn -> Req.get(url, receive_timeout: 5_000) end)
  end
end

For consumer groups with multiple partitions, use a single shared heartbeat URL:

# lib/my_app/consumer_group.ex
defmodule MyApp.ConsumerGroup do
  use KafkaEx.GenConsumer.Supervisor,
    consumer_module: MyApp.Consumer,
    group_name: "my-app-consumers",
    topics: ["events", "commands"]
end

Scheduled pipeline heartbeat

Add a separate heartbeat for when consumer traffic is genuinely low (no messages to process):

# lib/my_app/pipeline_monitor.ex
defmodule MyApp.PipelineMonitor do
  use GenServer

  require Logger

  @check_interval :timer.minutes(5)

  def start_link(_), do: GenServer.start_link(__MODULE__, %{}, name: __MODULE__)

  def init(state) do
    schedule_check()
    {:ok, state}
  end

  def handle_info(:check, state) do
    # Verify consumer group is registered and active
    case KafkaEx.consumer_group_exists?("my-app-consumers") do
      true ->
        ping_pipeline_heartbeat()
      false ->
        Logger.error("Consumer group not found — pipeline may have stopped")
    end

    schedule_check()
    {:noreply, state}
  end

  defp schedule_check, do: Process.send_after(self(), :check, @check_interval)

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

Configure heartbeat URLs:

# config/runtime.exs
config :my_app, :vigilmon,
  consumer_heartbeat_url: System.get_env("VIGILMON_CONSUMER_HEARTBEAT_URL"),
  pipeline_heartbeat_url: System.get_env("VIGILMON_PIPELINE_HEARTBEAT_URL")

In Vigilmon, create Heartbeat Monitors:

  1. Consumer Heartbeat: expected interval 2 minutes (fires on every message batch)
  2. Pipeline Heartbeat: expected interval 6 minutes (fires on schedule)

If message batches stop arriving and the pipeline monitor stops pinging, both heartbeats miss and Vigilmon alerts you.

Step 4: Alerts via Slack

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

To create a webhook:

  1. api.slack.com/apps → Create New App → From scratch
  2. Enable Incoming Webhooks → Add New Webhook
  3. Pick your data pipeline alerts channel and copy the URL

Enable the Slack channel on all monitors. When the consumer group stops processing:

🔴 HEARTBEAT MISSED: Kafka Consumer Pipeline
Expected ping every 2 minutes
Last ping: 18 minutes ago

🔴 DOWN: yourdomain.com/health
Keyword check failed: "kafka_broker":"ok" not found
Detected from: EU-West, US-East

For event-driven architectures, a stalled consumer pipeline can have cascading effects — delayed order processing, failed notifications, or inconsistent read models.

Step 5: Monitor the Kafka broker directly

Add a TCP monitor for the Kafka broker port to distinguish broker failures from application failures:

  1. Click New Monitor → TCP
  2. Host: your Kafka broker hostname
  3. Port: 9092
  4. Interval: 1 minute

This separates:

  • Broker down (TCP monitor fails): Kafka itself is unreachable
  • Consumer crashed (HTTP/heartbeat monitor fails, TCP monitor passes): your KafkaEx application has an issue

For a multi-broker cluster, add one TCP monitor per broker. A partial broker failure that causes partition leadership changes will show up as individual TCP monitors failing while others pass.

Step 6: Status page and badge

Status page:

  1. Go to Status Pages → New Status Page in Vigilmon
  2. Add your HTTP monitor, consumer heartbeat, pipeline heartbeat, and TCP broker monitors
  3. Copy the public URL

Share with your data engineering team so everyone sees pipeline health at a glance without digging through logs.

README badge:

![Kafka Pipeline Uptime](https://vigilmon.online/badge/your-monitor-slug)

What you've built

| What | How | |------|-----| | Health check endpoint | Plug that tests broker connectivity + consumer liveness | | Broker connectivity check | KafkaEx.metadata/1 liveness probe | | HTTP uptime monitoring | Vigilmon HTTP monitor → /health | | Keyword check | Verifies "kafka_broker":"ok" in response | | Consumer message heartbeat | Ping on each successful handle_message_set/2 batch | | Scheduled pipeline heartbeat | Periodic ping via PipelineMonitor GenServer | | Direct broker TCP check | Vigilmon TCP monitor on port 9092 per broker | | Slack alerts | Vigilmon Slack notification channel | | Status page | Vigilmon public status page |

BEAM keeps your KafkaEx processes alive. Vigilmon tells you when messages have actually stopped flowing.

Next steps

  • Expose consumer group lag metrics in your health response (query KafkaEx.offset_fetch/2 vs latest offset)
  • Use Vigilmon's response time history to correlate broker latency spikes with consumer throughput drops
  • Add separate heartbeat monitors per Kafka topic for fine-grained visibility into which pipelines are healthy
  • If you use GenStage or Broadway over KafkaEx, add heartbeats at each stage boundary to pinpoint where backpressure accumulates

Get started free at vigilmon.online.

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →
← Back to all articles