How to Monitor Apache Kafka with Vigilmon

Apache Kafka is the backbone of event-driven architectures worldwide. Kafka powers real-time data pipelines, event sourcing systems, and message queues for some of the world's largest applications. But Kafka cluster health and your Kafka consumer health are two different concerns — and both need monitoring.

This guide shows how to use Vigilmon to monitor services that depend on Kafka, catching failures before they cascade.

The Kafka Monitoring Challenge

Kafka itself exposes metrics via JMX and the Kafka Metrics API, which tools like Prometheus + Grafana consume. But those internal metrics don't give you the user-facing view: is your application still processing events?

Vigilmon solves the complementary problem: monitoring your Kafka-consuming application's health endpoint, which validates the full stack — Kafka connectivity, consumer group lag, and application health — from the outside.

Common Kafka-Related Failure Modes

• Broker unavailability: One or more Kafka brokers go down
• Topic deletion: A required topic is accidentally deleted
• Consumer group rebalancing: Temporary processing halt during rebalance
• Consumer lag buildup: Producer outpaces consumers
• Authentication failures: SSL/SASL credentials expire or change
• Application crashes: The consumer process itself dies
• Network partitions: Consumer loses connectivity to brokers

Step 1: Add a Health Endpoint That Validates Kafka Connectivity

Node.js (kafkajs)

const { Kafka } = require('kafkajs');
const express = require('express');

const app = express();
const kafka = new Kafka({
  clientId: 'my-service',
  brokers: process.env.KAFKA_BROKERS.split(','),
});

const admin = kafka.admin();

app.get('/health', async (req, res) => {
  try {
    // Check Kafka broker connectivity
    const topics = await admin.listTopics();
    res.json({
      status: 'ok',
      kafka: 'connected',
      topics: topics.length
    });
  } catch (err) {
    res.status(503).json({
      status: 'error',
      kafka: err.message
    });
  }
});

Python (kafka-python)

from kafka import KafkaAdminClient
from kafka.errors import KafkaError
from flask import Flask, jsonify
import os

app = Flask(__name__)

def check_kafka():
    try:
        admin = KafkaAdminClient(
            bootstrap_servers=os.environ['KAFKA_BROKERS'],
            client_id='health-check'
        )
        topics = admin.list_topics()
        admin.close()
        return True, len(topics)
    except KafkaError as e:
        return False, str(e)

@app.route('/health')
def health():
    ok, info = check_kafka()
    if ok:
        return jsonify({'status': 'ok', 'kafka': 'connected', 'topics': info})
    return jsonify({'status': 'error', 'kafka': info}), 503

Java / Spring Boot

@RestController
public class HealthController {

    @Autowired
    private KafkaAdmin kafkaAdmin;

    @GetMapping("/health")
    public ResponseEntity<Map<String, Object>> health() {
        Map<String, Object> response = new HashMap<>();
        try (AdminClient client = AdminClient.create(kafkaAdmin.getConfigurationProperties())) {
            ListTopicsResult result = client.listTopics();
            Set<String> topics = result.names().get(5, TimeUnit.SECONDS);
            response.put("status", "ok");
            response.put("kafka", "connected");
            response.put("topics", topics.size());
            return ResponseEntity.ok(response);
        } catch (Exception e) {
            response.put("status", "error");
            response.put("kafka", e.getMessage());
            return ResponseEntity.status(503).body(response);
        }
    }
}

Go (confluent-kafka-go)

package main

import (
    "encoding/json"
    "net/http"
    "github.com/confluentinc/confluent-kafka-go/kafka"
)

func healthHandler(w http.ResponseWriter, r *http.Request) {
    admin, err := kafka.NewAdminClient(&kafka.ConfigMap{
        "bootstrap.servers": getenv("KAFKA_BROKERS"),
    })
    w.Header().Set("Content-Type", "application/json")
    if err != nil {
        w.WriteHeader(http.StatusServiceUnavailable)
        json.NewEncoder(w).Encode(map[string]string{"status": "error", "kafka": err.Error()})
        return
    }
    defer admin.Close()

    _, err = admin.GetMetadata(nil, true, 5000)
    if err != nil {
        w.WriteHeader(http.StatusServiceUnavailable)
        json.NewEncoder(w).Encode(map[string]string{"status": "error", "kafka": err.Error()})
        return
    }
    json.NewEncoder(w).Encode(map[string]string{"status": "ok", "kafka": "connected"})
}

Step 2: Test the Health Endpoint

curl https://yourservice.com/health
# When Kafka is healthy:
{"status":"ok","kafka":"connected","topics":12}

# When Kafka is unreachable:
# HTTP 503
{"status":"error","kafka":"Leader Not Available"}

Step 3: Configure Vigilmon

1. Log into vigilmon.online
2. Click New Monitor
3. Set:
   • URL: https://yourservice.com/health
   • Type: HTTP
   • Interval: 60 seconds
   • Expected status: 200
4. Add alert channels (email + Slack)
5. Save

Vigilmon will alert your team within 60 seconds if the Kafka connection fails.

Monitoring Consumer Lag

For critical Kafka consumers, also expose consumer lag in your health endpoint:

// In kafkajs
const admin = kafka.admin();
const offsets = await admin.fetchOffsets({ 
  groupId: 'my-consumer-group',
  topics: ['my-topic']
});

const topicOffsets = await admin.fetchTopicOffsets('my-topic');

// Calculate lag per partition
const lag = offsets[0].partitions.reduce((total, partition) => {
  const topicPartition = topicOffsets.find(p => p.partition === partition.partition);
  return total + (parseInt(topicPartition.offset) - parseInt(partition.offset));
}, 0);

if (lag > 10000) {
  // High lag = potential processing issue
  return res.status(503).json({ status: 'error', lag: lag, message: 'Consumer lag too high' });
}

This turns consumer lag into a health signal that Vigilmon can detect.

Alerting Strategy

For Kafka + Application Health

1. Vigilmon → monitors /health endpoint → alerts Slack + email
2. Prometheus + Alertmanager → monitors JMX metrics → alerts for broker-level issues
3. Kafka UI → human-readable cluster state for debugging

Vigilmon handles the external, user-facing health signal. Prometheus handles internal cluster metrics.

Summary

Kafka is powerful, but your Kafka-consuming services still need external monitoring. In 3 steps:

1. Add a /health endpoint that validates Kafka broker connectivity
2. Configure Vigilmon to monitor it every 60 seconds
3. Set up Slack/email alerts for instant notification when Kafka goes down

Start monitoring your Kafka-backed services with Vigilmon →