How to Monitor Apache Pulsar: Broker Health, Topic Backlog, and Consumer Lag with Vigilmon

Apache Pulsar is a cloud-native distributed messaging and streaming platform built for multi-tenancy and geo-replication. Unlike Kafka, Pulsar separates storage (BookKeeper) from compute (brokers), which gives you better scalability — but also more failure surfaces. A broker can be healthy while BookKeeper loses a ledger, or a geo-replication link can silently fall behind while the dashboard shows green.

Vigilmon gives you external visibility into your Pulsar cluster — broker HTTP health APIs, consumer lag via a sidecar, and heartbeat monitoring for consumer applications. This tutorial covers all three layers.

Why Pulsar Needs External Monitoring

Pulsar's built-in tooling (Pulsar Manager, Prometheus metrics, the admin REST API) is rich — but it only works if someone is watching the dashboards. External monitoring with Vigilmon adds:

• Proactive broker health alerts when the Pulsar admin API stops responding
• Topic backlog depth monitoring via a lightweight health sidecar
• Consumer lag tracking to catch stuck subscriptions before they become message loss incidents
• BookKeeper ledger health checks from outside the cluster perimeter
• Geo-replication lag alerts for multi-region deployments

These layers complement each other: a broker restart can cause backlog to spike even if the broker recovers, and a geo-replication failure can occur without any broker error.

Step 1: Expose Pulsar Health Endpoints

Pulsar's admin REST API is the best starting point. By default, it runs on port 8080 for the broker.

Broker Health Check

# Pulsar broker built-in health endpoint
curl -i http://localhost:8080/admin/v2/brokers/health

# Broker list (proxy-accessible)
curl -i http://localhost:8080/admin/v2/brokers/

# Topic stats (backlog check)
curl -i "http://localhost:8080/admin/v2/persistent/public/default/orders/stats"

For a production Pulsar cluster, these endpoints are typically behind a proxy or service mesh. Configure your Vigilmon monitor to hit the proxy URL.

Node.js Health Sidecar (Full Backlog and Lag Check)

// health/pulsar.js
const express = require('express');
const fetch = require('node-fetch');

const app = express();
const PULSAR_ADMIN = process.env.PULSAR_ADMIN_URL || 'http://localhost:8080';
const TENANT = process.env.PULSAR_TENANT || 'public';
const NAMESPACE = process.env.PULSAR_NAMESPACE || 'default';
const TOPICS = (process.env.PULSAR_TOPICS || '').split(',').filter(Boolean);
const BACKLOG_THRESHOLD = parseInt(process.env.PULSAR_BACKLOG_THRESHOLD || '10000');
const LAG_THRESHOLD = parseInt(process.env.PULSAR_LAG_THRESHOLD || '5000');

app.get('/health/pulsar', async (req, res) => {
  try {
    // 1. Broker health
    const brokerRes = await fetch(`${PULSAR_ADMIN}/admin/v2/brokers/health`, {
      signal: AbortSignal.timeout(5000),
    });
    if (!brokerRes.ok) {
      return res.status(503).json({ status: 'down', reason: 'broker_health_failed' });
    }

    // 2. Topic backlog check
    const issues = [];
    for (const topic of TOPICS) {
      const statsRes = await fetch(
        `${PULSAR_ADMIN}/admin/v2/persistent/${TENANT}/${NAMESPACE}/${topic}/stats`
      );
      if (!statsRes.ok) continue;
      const stats = await statsRes.json();

      // Aggregate backlog across all subscriptions
      for (const [subName, sub] of Object.entries(stats.subscriptions || {})) {
        if (sub.msgBacklog > BACKLOG_THRESHOLD) {
          issues.push({ topic, subscription: subName, backlog: sub.msgBacklog });
        }
      }
    }

    if (issues.length > 0) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'backlog_threshold_exceeded',
        issues,
      });
    }

    return res.status(200).json({ status: 'ok', topics_checked: TOPICS.length });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

// BookKeeper health check
app.get('/health/pulsar/bookkeeper', async (req, res) => {
  try {
    const BOOKIE_HTTP = process.env.BOOKIE_HTTP_URL || 'http://localhost:8082';
    const bkRes = await fetch(`${BOOKIE_HTTP}/api/v1/bookie/is_ready`, {
      signal: AbortSignal.timeout(5000),
    });
    if (!bkRes.ok) {
      return res.status(503).json({ status: 'down', reason: 'bookkeeper_not_ready' });
    }
    return res.status(200).json({ status: 'ok' });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

// Geo-replication lag check
app.get('/health/pulsar/geo', async (req, res) => {
  try {
    const REMOTE_CLUSTER = process.env.PULSAR_REMOTE_CLUSTER;
    if (!REMOTE_CLUSTER) return res.status(200).json({ status: 'skipped', reason: 'no_remote_cluster' });

    const statsRes = await fetch(
      `${PULSAR_ADMIN}/admin/v2/persistent/${TENANT}/${NAMESPACE}/${TOPICS[0]}/stats`
    );
    if (!statsRes.ok) return res.status(503).json({ status: 'down' });
    const stats = await statsRes.json();

    const repStats = stats.replication?.[REMOTE_CLUSTER];
    if (!repStats) return res.status(503).json({ status: 'degraded', reason: 'no_replication_stats' });

    const LAG_LIMIT = parseInt(process.env.GEO_LAG_THRESHOLD || '1000');
    if (repStats.replicationBacklog > LAG_LIMIT) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'geo_replication_lag',
        cluster: REMOTE_CLUSTER,
        backlog: repStats.replicationBacklog,
      });
    }

    return res.status(200).json({ status: 'ok', cluster: REMOTE_CLUSTER });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

app.listen(3007, () => console.log('Pulsar health sidecar running on :3007'));

Step 2: Configure Vigilmon HTTP Monitors for Pulsar

1. Log in to vigilmon.online and go to Monitors → New Monitor
2. Choose HTTP / HTTPS
3. Set the URL to your broker health endpoint: https://pulsar-proxy.example.com/admin/v2/brokers/health
4. Set check interval to 1 minute
5. Under Expected response:
   • Status code: 200
   • Response time threshold: 5000ms
6. Assign your PagerDuty or Slack alert channel
7. Save the monitor

Add additional monitors for each health layer:

| Monitor | URL | Expected | Priority | |---|---|---|---| | Broker health | /health/pulsar | 200, body "status":"ok" | P1 | | BookKeeper | /health/pulsar/bookkeeper | 200 | P1 | | Topic backlog | /health/pulsar (backlog check) | 200, no issues | P2 | | Geo-replication | /health/pulsar/geo | 200, no lag | P2 |

Use separate monitors for broker and BookKeeper — they can fail independently.

Step 3: Heartbeat Monitoring for Pulsar Consumers

A consumer application can fall silent while the broker is healthy. Vigilmon heartbeat monitors detect this by expecting a periodic ping from your consumer. If the ping stops, Vigilmon fires an alert.

Set Up the Heartbeat Monitor

1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
2. Name: pulsar-orders-consumer
3. Expected interval: 5 minutes
4. Grace period: 10 minutes
5. Save and copy the heartbeat URL: https://vigilmon.online/heartbeat/abc123xyz

Wire the Heartbeat Into Your Consumer (Node.js)

// consumer.js
const Pulsar = require('pulsar-client');
const fetch = require('node-fetch');

const client = new Pulsar.Client({
  serviceUrl: process.env.PULSAR_SERVICE_URL || 'pulsar://localhost:6650',
});

let processedSinceHeartbeat = 0;
const HEARTBEAT_INTERVAL_MS = 60_000;

async function run() {
  const consumer = await client.subscribe({
    topic: `persistent://public/default/${process.env.PULSAR_TOPIC}`,
    subscription: process.env.PULSAR_SUBSCRIPTION,
    subscriptionType: 'Shared',
  });

  // Send heartbeat on a timer — decoupled from message throughput
  setInterval(async () => {
    try {
      await fetch(process.env.VIGILMON_HEARTBEAT_URL, { signal: AbortSignal.timeout(5000) });
    } catch (_) {}
  }, HEARTBEAT_INTERVAL_MS);

  while (true) {
    const msg = await consumer.receive();
    try {
      await processMessage(msg);
      await consumer.acknowledge(msg);
    } catch (err) {
      await consumer.negativeAcknowledge(msg);
    }
  }
}

run();

Python Consumer (pulsar-client)

import pulsar, requests, os, threading, time

client = pulsar.Client(os.environ['PULSAR_SERVICE_URL'])
consumer = client.subscribe(
    f"persistent://public/default/{os.environ['PULSAR_TOPIC']}",
    subscription_name=os.environ['PULSAR_SUBSCRIPTION'],
)

def send_heartbeat():
    while True:
        try:
            requests.get(os.environ['VIGILMON_HEARTBEAT_URL'], timeout=5)
        except Exception:
            pass
        time.sleep(60)

threading.Thread(target=send_heartbeat, daemon=True).start()

while True:
    msg = consumer.receive()
    try:
        process_message(msg)
        consumer.acknowledge(msg)
    except Exception:
        consumer.negative_acknowledge(msg)

Step 4: Monitoring Geo-Replication and SLA Compliance

For multi-region Pulsar deployments, geo-replication failures are among the hardest to detect. Messages continue flowing to the local cluster while the remote cluster falls behind — often silently.

Configure Vigilmon to probe your geo-replication health endpoint every minute. Set response time thresholds:

• Alert at 2000ms for the broker health endpoint
• Alert at 10 000ms for the geo-replication lag check (allow for cross-cluster latency)

For SLA dashboards, use Vigilmon's uptime reports to track messaging availability over 30/90-day windows and share them with stakeholders as your messaging SLA evidence.

Summary

Apache Pulsar's layered architecture — brokers, BookKeeper, geo-replication — requires layered monitoring:

| Monitor Type | What It Covers | |---|---| | HTTP: broker health | Broker availability, REST API responsiveness | | HTTP: BookKeeper | Ledger storage availability | | HTTP: backlog check | Topic subscription backlog depth | | HTTP: geo-replication | Cross-cluster replication lag | | Heartbeat | Consumer application liveness |

Get started free at vigilmon.online — your first Pulsar monitor is live in under two minutes.