Uptime Monitoring for Apache Kafka Applications (Free, Multi-Region)

Kafka is designed to be resilient — replication, leader election, and partition failover happen automatically. But "Kafka is running" and "your Kafka application is healthy" are two different things.

Your producer might be silently dropping messages. Your consumer might be hours behind with no one noticing. A broker restart during a network partition can leave your consumer group in a rebalancing loop. This guide shows you how to detect these failures before they cascade.

---

The failure modes that Kafka's built-in resilience doesn't catch

Consumer lag — Your consumer is connected but processing too slowly. Messages queue up. What was a 200ms pipeline becomes a 4-hour pipeline. Kafka reports the consumer as "alive" because it's sending heartbeats. The business logic is broken.

Silent producer drops — Your producer sends a message, Kafka returns an ack, but the message went to a dead-letter or was filtered by a schema validator you forgot to update. You think you're producing; nothing is consuming what you produced.

Rebalancing loops — A consumer group in constant rebalance never makes progress. Members join, get assigned partitions, then leave before they can commit offsets. Kafka considers this "normal operations"; your application is stuck.

Connectivity from app to broker — A firewall rule change or misconfigured SASL credential means your producer can't reach the broker. Your application logs Error connecting to node but nothing alerts you.

---

Step 1: Add a Kafka health check to your application

Node.js (KafkaJS)

// health/kafka.ts
import { Kafka, Admin } from 'kafkajs'

const kafka = new Kafka({
  brokers: (process.env.KAFKA_BROKERS ?? 'localhost:9092').split(','),
  ssl: process.env.KAFKA_SSL === 'true',
  sasl: process.env.KAFKA_USERNAME ? {
    mechanism: 'plain',
    username: process.env.KAFKA_USERNAME,
    password: process.env.KAFKA_PASSWORD!,
  } : undefined,
})

let admin: Admin | null = null

async function getAdmin(): Promise<Admin> {
  if (!admin) {
    admin = kafka.admin()
    await admin.connect()
  }
  return admin
}

export async function checkKafka(): Promise<{
  status: 'ok' | 'error' | 'warn'
  latencyMs?: number
  brokerCount?: number
  error?: string
}> {
  const start = Date.now()
  try {
    const a = await getAdmin()
    const metadata = await a.describeCluster()
    return {
      status: 'ok',
      latencyMs: Date.now() - start,
      brokerCount: metadata.brokers.length,
    }
  } catch (err: any) {
    return { status: 'error', error: err.message }
  }
}

export async function checkConsumerLag(
  groupId: string,
  topics: string[]
): Promise<{ status: string; totalLag: number; partitions: any[] }> {
  try {
    const a = await getAdmin()
    const offsets = await a.fetchOffsets({ groupId, topics })
    const topicOffsets = await a.fetchTopicOffsets(topics[0])

    let totalLag = 0
    const partitions = []

    for (const topic of offsets) {
      for (const partition of topic.partitions) {
        const topicEnd = topicOffsets.find(
          (t: any) => t.partition === partition.partition
        )
        if (topicEnd) {
          const lag = parseInt(topicEnd.offset) - parseInt(partition.offset)
          totalLag += lag
          partitions.push({
            topic: topic.topic,
            partition: partition.partition,
            lag,
            consumerOffset: partition.offset,
            latestOffset: topicEnd.offset,
          })
        }
      }
    }

    const LAG_WARN_THRESHOLD = 1000
    const LAG_CRIT_THRESHOLD = 10000

    return {
      status: totalLag > LAG_CRIT_THRESHOLD ? 'critical' : totalLag > LAG_WARN_THRESHOLD ? 'warn' : 'ok',
      totalLag,
      partitions,
    }
  } catch (err: any) {
    return { status: 'error', totalLag: -1, partitions: [], error: err.message } as any
  }
}

---

Step 2: Expose it in your health endpoint

// routes/health.ts
import { checkKafka, checkConsumerLag } from '../health/kafka'

app.get('/health', async (req, res) => {
  const [kafka, lag] = await Promise.all([
    checkKafka(),
    checkConsumerLag(
      process.env.KAFKA_CONSUMER_GROUP ?? 'default-group',
      (process.env.KAFKA_TOPICS ?? 'events').split(',')
    ),
  ])

  const isOk = kafka.status === 'ok' && lag.status === 'ok'

  res.status(isOk ? 200 : 503).json({
    status: isOk ? 'ok' : 'degraded',
    checks: {
      kafka_broker: kafka,
      consumer_lag: lag,
    },
    timestamp: new Date().toISOString(),
  })
})

---

Step 3: Python version (confluent-kafka)

# health/kafka.py
from confluent_kafka.admin import AdminClient
from confluent_kafka import Consumer, TopicPartition
import os
import time

def check_kafka_broker() -> dict:
    start = time.time()
    try:
        admin = AdminClient({
            'bootstrap.servers': os.environ.get('KAFKA_BROKERS', 'localhost:9092'),
            'socket.timeout.ms': 3000,
        })
        metadata = admin.list_topics(timeout=3)
        return {
            'status': 'ok',
            'latency_ms': round((time.time() - start) * 1000),
            'topic_count': len(metadata.topics),
        }
    except Exception as e:
        return {'status': 'error', 'error': str(e)}

def check_consumer_lag(group_id: str, topics: list) -> dict:
    try:
        consumer = Consumer({
            'bootstrap.servers': os.environ.get('KAFKA_BROKERS', 'localhost:9092'),
            'group.id': group_id,
        })
        committed = consumer.committed([TopicPartition(t, 0) for t in topics])
        # ... check watermarks vs committed offsets
        consumer.close()
        return {'status': 'ok', 'checked_topics': topics}
    except Exception as e:
        return {'status': 'error', 'error': str(e)}

---

Step 4: Set up consumer lag alerting thresholds

Lag Level	Messages Behind	Action
Normal	0 – 1,000	No action
Warning	1,000 – 10,000	Investigate consumer throughput
Critical	> 10,000	Page on-call, consider scaling consumers

Return HTTP 200 for normal, 200 with a warning body for warn, and 503 for critical/error. Your monitor checks status codes; your ops team checks the body.

---

Step 5: External monitoring setup

1. Go to vigilmon.online — free tier.
2. Create an HTTP(S) monitor for https://your-app.com/health.
3. Interval: 60s
4. Expected status: 200
5. Regions: 2+ (especially important for Kafka — regional network changes are a common cause of broker connectivity failures)
6. Alert: email + Slack

---

Recap

1. Check broker connectivity via AdminClient.describeCluster() — not just "can I connect to the port."
2. Check consumer lag separately — a connected consumer that's 10,000 messages behind is functionally broken.
3. Expose broker status and lag in a single /health endpoint with structured JSON.
4. Set lag thresholds: warning at 1k messages, critical at 10k.
5. External monitoring at vigilmon.online catches application-layer issues Kafka's own tooling doesn't surface.

Kafka handles the hard distributed systems problems. You handle knowing when your producers and consumers have stopped working.