tutorial

How to Monitor ksqlDB with Vigilmon

ksqlDB persistent queries can fail or stall silently, leaving your streaming SQL pipelines dead without any visible error. Learn how to monitor ksqlDB query health, REST API availability, and processing lag with Vigilmon HTTP and heartbeat monitors.

ksqlDB brings SQL semantics to Apache Kafka, letting you create persistent streaming queries, materialized views, and real-time aggregations without writing Java code. But ksqlDB's operational surface is deceptively quiet: a persistent query can enter a ERROR state, a pull query can time out, or consumer lag can climb for an hour before anyone notices — because ksqlDB does not page you when this happens.

Vigilmon provides external monitoring for ksqlDB through HTTP endpoint monitoring of the ksqlDB REST API and heartbeat monitoring for downstream consumers of ksqlDB output topics. This tutorial walks through both approaches so you know the moment your streaming SQL pipeline degrades.

Why ksqlDB Needs External Monitoring

ksqlDB exposes a REST API and a metrics endpoint, but these are passive — they report state only when queried. External monitoring with Vigilmon adds:

  • Continuous probing of ksqlDB REST API availability every minute
  • Persistent query state checking — a query in ERROR state returns wrong data silently
  • Processing lag tracking via the ksqlDB status endpoint
  • Downstream heartbeat monitoring to catch cases where ksqlDB is running but output topics are stale
  • Multi-region external verification independent of your internal metrics pipeline

ksqlDB failures often look like data freshness issues rather than service failures. By the time users complain about stale dashboards, the persistent query may have been in ERROR for hours.

Step 1: Probe the ksqlDB REST API

ksqlDB exposes a REST API on port 8088 by default. You can probe it directly or via a thin health proxy.

Direct REST API Health Check

ksqlDB's /info endpoint returns server status:

# Test locally
curl -s http://localhost:8088/info | jq .

Expected healthy response:

{
  "KsqlServerInfo": {
    "version": "7.6.0",
    "kafkaClusterId": "abc123",
    "ksqlServiceId": "default_",
    "serverStatus": "RUNNING"
  }
}

Point your Vigilmon monitor directly at this endpoint if ksqlDB is externally reachable. If it runs inside a private network, add a health proxy.

Persistent Query Health Proxy (Node.js)

ksqlDB's /info only tells you the server is running. A more valuable check queries persistent query states:

// ksqldb-health.js
const express = require('express');
const axios = require('axios');

const app = express();
const KSQLDB_URL = process.env.KSQLDB_URL || 'http://localhost:8088';

app.get('/health/ksqldb', async (req, res) => {
  try {
    // Check server info
    const infoRes = await axios.get(`${KSQLDB_URL}/info`, { timeout: 5000 });
    const serverStatus = infoRes.data?.KsqlServerInfo?.serverStatus;

    if (serverStatus !== 'RUNNING') {
      return res.status(503).json({
        status: 'down',
        reason: 'server_not_running',
        serverStatus,
      });
    }

    // List persistent queries and check for errors
    const queriesRes = await axios.post(
      `${KSQLDB_URL}/ksql`,
      { ksql: 'LIST QUERIES;' },
      { headers: { 'Content-Type': 'application/vnd.ksql.v1+json' }, timeout: 8000 }
    );

    const queries = queriesRes.data?.[0]?.queries || [];
    const erroredQueries = queries.filter(q => q.state === 'ERROR');
    const runningQueries = queries.filter(q => q.state === 'RUNNING');

    if (erroredQueries.length > 0) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'queries_in_error_state',
        erroredQueries: erroredQueries.map(q => q.id),
        totalQueries: queries.length,
      });
    }

    return res.status(200).json({
      status: 'ok',
      serverStatus,
      runningQueries: runningQueries.length,
      totalQueries: queries.length,
    });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

app.listen(3005);

Python Health Proxy (FastAPI)

# ksqldb_health.py
from fastapi import FastAPI
from fastapi.responses import JSONResponse
import httpx, os

app = FastAPI()
KSQLDB_URL = os.environ.get("KSQLDB_URL", "http://localhost:8088")

@app.get("/health/ksqldb")
async def ksqldb_health():
    async with httpx.AsyncClient(timeout=8.0) as client:
        try:
            # Check server info
            info = await client.get(f"{KSQLDB_URL}/info")
            server_status = info.json().get("KsqlServerInfo", {}).get("serverStatus")

            if server_status != "RUNNING":
                return JSONResponse(status_code=503, content={
                    "status": "down", "serverStatus": server_status
                })

            # Check persistent query states
            queries_resp = await client.post(
                f"{KSQLDB_URL}/ksql",
                json={"ksql": "LIST QUERIES;"},
                headers={"Content-Type": "application/vnd.ksql.v1+json"}
            )
            queries = queries_resp.json()[0].get("queries", [])
            errored = [q["id"] for q in queries if q["state"] == "ERROR"]

            if errored:
                return JSONResponse(status_code=503, content={
                    "status": "degraded",
                    "reason": "queries_in_error_state",
                    "erroredQueries": errored,
                })

            return {"status": "ok", "runningQueries": len(queries)}

        except Exception as e:
            return JSONResponse(status_code=503, content={"status": "down", "error": str(e)})

Step 2: Monitor Consumer Lag for ksqlDB Output Topics

ksqlDB writes query results to Kafka topics. Even when queries run, output topic lag can indicate that downstream consumers are not keeping up. Add a lag-aware endpoint:

// lag endpoint added to ksqldb-health.js
const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'ksqldb-lag-monitor',
  brokers: (process.env.KAFKA_BROKERS || 'localhost:9092').split(','),
});
const admin = kafka.admin();

app.get('/health/ksqldb/output-lag', async (req, res) => {
  const outputTopics = (process.env.KSQLDB_OUTPUT_TOPICS || '').split(',').filter(Boolean);
  const consumerGroup = process.env.KSQLDB_CONSUMER_GROUP || '_confluent-ksql-default_query_CSAS_ORDERS_ENRICHED_0';
  const lagThreshold = parseInt(process.env.LAG_THRESHOLD || '10000');

  try {
    await admin.connect();

    const groupOffsets = await admin.fetchOffsets({ groupId: consumerGroup, topics: outputTopics });
    let maxLag = 0;

    for (const topic of outputTopics) {
      const topicOffsets = await admin.fetchTopicOffsets(topic);
      const consumerOffsets = groupOffsets.filter(o => o.topic === topic);
      for (const partition of topicOffsets) {
        const consumed = consumerOffsets.find(o => o.partition === partition.partition);
        if (consumed) {
          const lag = parseInt(partition.high) - parseInt(consumed.offset);
          maxLag = Math.max(maxLag, lag);
        }
      }
    }

    await admin.disconnect();

    if (maxLag > lagThreshold) {
      return res.status(503).json({ status: 'degraded', reason: 'output_lag_exceeded', maxLag });
    }
    return res.status(200).json({ status: 'ok', maxLag });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

Step 3: Configure Vigilmon HTTP Monitors

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. Set the URL to your ksqlDB health proxy: https://your-app.example.com/health/ksqldb
  4. Set the check interval to 1 minute
  5. Under Expected response, configure:
    • Status code: 200
    • Response body contains: "status":"ok"
    • Response time threshold: 10000ms (ksqlDB query listing can be slow under load)
  6. Under Alert channels, assign your Slack or PagerDuty channel
  7. Save the monitor

Add a second monitor for output lag:

  • URL: https://your-app.example.com/health/ksqldb/output-lag
  • Expected: 200, body contains "status":"ok"
  • Interval: 1 minute
  • Alert channel: data engineering Slack channel

If your ksqlDB REST API is directly reachable (Confluent Cloud or internal), you can also monitor /info directly:

  • URL: https://ksqldb.example.com/info
  • Expected: 200, response body contains: "serverStatus":"RUNNING"
  • Interval: 30 seconds

Step 4: Heartbeat Monitoring for ksqlDB Output Consumers

ksqlDB output topics are consumed by downstream services — dashboards, data warehouses, notification systems. Even if ksqlDB is running, a downstream consumer stalling means real users see stale data.

Vigilmon heartbeat monitors catch this gap: your downstream consumer pings Vigilmon after each batch of processed ksqlDB output records.

Set Up the Heartbeat Monitor

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Set the name: ksqldb-orders-enriched-consumer
  3. Set the expected interval: 5 minutes
  4. Set the grace period: 10 minutes
  5. Save — copy the heartbeat URL: https://vigilmon.online/heartbeat/abc123xyz

Wire Heartbeat Into Your Downstream Consumer

// ksqldb-output-consumer.js
const { Kafka } = require('kafkajs');
const axios = require('axios');

const kafka = new Kafka({
  clientId: 'orders-enriched-consumer',
  brokers: process.env.KAFKA_BROKERS.split(','),
});

const consumer = kafka.consumer({ groupId: 'orders-enriched-dashboard-group' });

let recordCount = 0;
const HEARTBEAT_EVERY = 50;

async function run() {
  await consumer.connect();
  await consumer.subscribe({ topic: 'ORDERS_ENRICHED', fromBeginning: false });

  await consumer.run({
    eachMessage: async ({ message }) => {
      await processEnrichedOrder(JSON.parse(message.value.toString()));
      recordCount++;

      if (recordCount % HEARTBEAT_EVERY === 0) {
        await axios.get(process.env.VIGILMON_HEARTBEAT_URL).catch(() => {});
      }
    },
  });
}

run();

For Python consumers of ksqlDB output:

# ksqldb_consumer.py
from confluent_kafka import Consumer
import requests, os, time

consumer = Consumer({
    'bootstrap.servers': os.environ['KAFKA_BROKERS'],
    'group.id': 'orders-enriched-dashboard-group',
    'auto.offset.reset': 'latest',
})
consumer.subscribe(['ORDERS_ENRICHED'])

last_heartbeat = time.time()
HEARTBEAT_INTERVAL = 60

try:
    while True:
        msg = consumer.poll(1.0)
        if msg and not msg.error():
            process_enriched_order(msg.value())
            if time.time() - last_heartbeat > HEARTBEAT_INTERVAL:
                requests.get(os.environ['VIGILMON_HEARTBEAT_URL'], timeout=5)
                last_heartbeat = time.time()
finally:
    consumer.close()

Step 5: Alert Routing for ksqlDB Failures

ksqlDB failures have different severities depending on which component fails:

| Monitor | Alert Channel | Priority | |---|---|---| | /health/ksqldb (server down or query ERROR) | Slack + PagerDuty | P1 | | /health/ksqldb/output-lag (lag exceeded) | Slack | P2 | | Heartbeat: orders-enriched-consumer | Slack + email | P2 | | Heartbeat: analytics-consumer | Email | P3 |

Set a response time alert threshold of 8000ms on the ksqlDB health endpoint. A slow LIST QUERIES response often precedes query state degradation by several minutes — it gives you lead time before a full failure.

For environments where ksqlDB query restarts cause brief REBALANCING → RUNNING transitions, configure a confirmation window in Vigilmon (2-3 failed checks before alerting) to avoid false alarms during rolling restarts.

Summary

ksqlDB makes streaming SQL simple but silent failures easy. External monitoring with Vigilmon gives you the observability layer that embedded metrics alone cannot provide:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /health/ksqldb | Server status, persistent query ERROR states | | HTTP monitor on /health/ksqldb/output-lag | Output topic consumer lag threshold breaches | | Heartbeat monitor | Downstream consumer liveness, data freshness proof |

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

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