How to Monitor Your Apache Kafka Topics with Vigilmon
Apache Kafka has become the backbone of event-driven architectures at scale — handling billions of messages per day for companies like LinkedIn, Uber, and Netflix. But Kafka's distributed nature makes it notoriously difficult to monitor: consumer lag, broker unavailability, partition leadership elections, and replication failures can all degrade your system without immediately visible external errors.
This guide shows you how to integrate Kafka health monitoring into Vigilmon.
Kafka Monitoring Basics
The three most important Kafka metrics to watch:
- Consumer group lag — how far behind consumers are from the latest message. Lag growing unboundedly = consumers can't keep up.
- Broker availability — is at least one Kafka broker reachable and acting as controller?
- Topic replication health — are all partition replicas in-sync?
Unlike HTTP services, Kafka doesn't expose a native HTTP health API. You need to build a bridge.
Step 1: Build a Kafka Health HTTP Endpoint
Node.js (kafkajs) example:
const { Kafka } = require('kafkajs');
const express = require('express');
const app = express();
const kafka = new Kafka({
clientId: 'health-checker',
brokers: process.env.KAFKA_BROKERS.split(','),
connectionTimeout: 3000,
requestTimeout: 5000
});
app.get('/health/kafka', async (req, res) => {
const admin = kafka.admin();
try {
await admin.connect();
// Check cluster metadata
const cluster = await admin.describeCluster();
const brokerCount = cluster.brokers.length;
// Check if controller is elected
if (!cluster.controller || cluster.controller === -1) {
return res.status(503).json({
status: 'no_controller',
brokers: brokerCount
});
}
await admin.disconnect();
res.json({
status: 'ok',
brokers: brokerCount,
controller: cluster.controller
});
} catch (err) {
try { await admin.disconnect(); } catch {}
res.status(503).json({ status: 'error', message: err.message });
}
});
Python (confluent-kafka) example:
from fastapi import FastAPI, Response
from confluent_kafka.admin import AdminClient
from confluent_kafka import KafkaException
import json
app = FastAPI()
@app.get("/health/kafka")
async def health_kafka():
admin = AdminClient({'bootstrap.servers': 'localhost:9092',
'socket.timeout.ms': 3000})
try:
metadata = admin.list_topics(timeout=5)
broker_count = len(metadata.brokers)
if broker_count == 0:
return Response(
content=json.dumps({"status": "no_brokers"}),
status_code=503
)
return {"status": "ok", "brokers": broker_count}
except KafkaException as e:
return Response(content=str(e), status_code=503)
Step 2: Monitor Consumer Group Lag
Consumer lag is the most operationally critical Kafka metric:
app.get('/health/kafka/lag', async (req, res) => {
const admin = kafka.admin();
try {
await admin.connect();
const groupId = 'my-consumer-group';
const topic = 'events';
// Get latest offsets
const topicOffsets = await admin.fetchTopicOffsets(topic);
// Get consumer group offsets
const groupOffsets = await admin.fetchOffsets({
groupId,
topics: [{ topic }]
});
let totalLag = 0;
for (const partition of topicOffsets) {
const groupPartition = groupOffsets[0].partitions
.find(p => p.partition === partition.partition);
if (groupPartition) {
const lag = parseInt(partition.offset) - parseInt(groupPartition.offset);
totalLag += Math.max(0, lag);
}
}
await admin.disconnect();
if (totalLag > 100000) {
return res.status(503).json({ status: 'high_lag', total_lag: totalLag });
}
res.json({ status: 'ok', consumer_lag: totalLag });
} catch (err) {
try { await admin.disconnect(); } catch {}
res.status(503).json({ status: 'error', message: err.message });
}
});
Step 3: Add Vigilmon Monitors
- Log in to vigilmon.online → Add Monitor
-
Monitor 1 — Broker health:
- URL:
https://your-app.com/health/kafka - Alert if: Status != 200
- URL:
-
Monitor 2 — Consumer lag:
- URL:
https://your-app.com/health/kafka/lag - Alert if: Status != 200 (triggered when lag > 100,000)
- URL:
- Interval: 60 seconds for both
Step 4: Consumer Process Heartbeat
For Kafka consumers (especially Flink jobs, Spark streaming, or custom consumer groups), add a heartbeat:
// Java Kafka consumer with Vigilmon heartbeat
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.URI;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
// In your consumer setup
ScheduledExecutorService heartbeatScheduler = Executors.newScheduledThreadPool(1);
heartbeatScheduler.scheduleAtFixedRate(() -> {
try {
HttpClient.newHttpClient().send(
HttpRequest.newBuilder()
.uri(URI.create("https://hb.vigilmon.online/YOUR-HEARTBEAT-ID"))
.GET().build(),
java.net.http.HttpResponse.BodyHandlers.ofString()
);
} catch (Exception e) {
// Log but don't crash consumer
}
}, 0, 60, TimeUnit.SECONDS);
Kafka Monitoring Coverage Table
| Monitor | Endpoint | Alerts On |
|---|---|---|
| Broker availability | /health/kafka |
No brokers reachable |
| Consumer lag | /health/kafka/lag |
Lag > 100,000 msgs |
| Consumer process | Vigilmon Heartbeat | Process died |
Common Kafka Failure Patterns
Consumer lag runaway: A downstream service became slow, causing consumers to process at 10% of normal throughput. Lag grew from 1,000 to 2,000,000 over 4 hours. Lag monitor would have fired at the 100,000 threshold within the first hour.
Controller election loop: A ZooKeeper instability caused Kafka controllers to cycle. Brokers were technically up but producing and consuming was failing. Health check returned 503; Vigilmon alerted within 60 seconds.
One-broker cluster loss: In a 3-broker cluster, a disk failure took broker 1 offline. Partitions with replicas on broker 1 became under-replicated. Cluster health check caught it immediately.
Conclusion
Kafka needs active health monitoring, not passive hope. By building HTTP health proxies for your Kafka cluster and connecting them to Vigilmon's multi-region monitoring, you get real-time alerts on broker availability, consumer lag, and partition health.
Top comments (0)