tutorial

How to Monitor Apache Flink: JobManager, Checkpoints, Backpressure, and Pipeline Uptime with Vigilmon

Apache Flink JobManager failures and checkpoint lag silently stall streaming pipelines. Learn how to monitor Flink availability, TaskManager slot utilization, checkpoint health, backpressure, and streaming SLAs with Vigilmon.

Apache Flink powers low-latency streaming pipelines — but when the JobManager goes down, all running jobs pause silently. Checkpoints stop completing. Backpressure builds on TaskManagers. Events pile up in Kafka topics. And unless you have external monitoring, your downstream consumers keep reading from stale state, completely unaware the pipeline is frozen.

Vigilmon gives you external visibility into Flink health through HTTP probe monitoring of the Flink REST API and heartbeat monitoring for streaming pipeline SLAs. This tutorial covers both.

Why Flink Needs External Monitoring

Flink's built-in web UI and metrics reporters (Prometheus, StatsD) provide deep internal visibility — but only when they're accessible and someone is watching. External monitoring with Vigilmon adds:

  • JobManager availability probing from outside your cluster network
  • Job failure detection via the Flink REST API job overview endpoint
  • Checkpoint lag alerting before checkpoint failures trigger job restarts
  • Backpressure detection to catch task bottlenecks before they cascade upstream
  • Heartbeat SLA monitoring so you know when a streaming job stops producing output events

These layers work together: Flink can show all jobs as "RUNNING" while checkpoint lag climbs toward the timeout threshold, and a job can checkpoint successfully while a downstream sink is silently dropping events.

Step 1: Build a Flink Health Endpoint

Flink exposes a REST API on the JobManager at http://jobmanager-host:8081. You need a sidecar that wraps critical checks into a single HTTP health response for Vigilmon to probe.

Node.js Health Sidecar

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

const app = express();
const FLINK_API = process.env.FLINK_API_URL || 'http://localhost:8081';

// JobManager overview — are all expected jobs running?
app.get('/health/flink', async (req, res) => {
  try {
    const overview = await axios.get(`${FLINK_API}/overview`, { timeout: 5000 });
    const data = overview.data;

    const failing = data['jobs-failed'] || 0;
    const cancelled = data['jobs-cancelled'] || 0;

    if (data.taskmanagers === 0) {
      return res.status(503).json({
        status: 'down',
        reason: 'no_taskmanagers',
        detail: data,
      });
    }

    if (failing > 0) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'failed_jobs',
        jobs_failed: failing,
      });
    }

    return res.status(200).json({
      status: 'ok',
      jobs_running: data['jobs-running'],
      taskmanagers: data.taskmanagers,
      slots_available: data['slots-available'],
      slots_total: data['slots-total'],
    });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

// Checkpoint health — check recent checkpoint completion lag
app.get('/health/flink/checkpoints', async (req, res) => {
  try {
    // List all running jobs
    const jobsRes = await axios.get(`${FLINK_API}/jobs`, { timeout: 5000 });
    const runningJobs = jobsRes.data.jobs.filter(j => j.status === 'RUNNING');

    const checkpointIssues = [];

    for (const job of runningJobs) {
      const cpRes = await axios.get(`${FLINK_API}/jobs/${job.id}/checkpoints`, { timeout: 5000 });
      const summary = cpRes.data.summary;
      const latest = cpRes.data.latest;

      if (!latest || !latest.completed) {
        checkpointIssues.push({ job: job.id, reason: 'no_completed_checkpoint' });
        continue;
      }

      // Alert if last checkpoint completed more than 10 minutes ago
      const lagMs = Date.now() - latest.completed['trigger-timestamp'];
      const lagThresholdMs = parseInt(process.env.CHECKPOINT_LAG_THRESHOLD_MS || '600000');
      if (lagMs > lagThresholdMs) {
        checkpointIssues.push({
          job: job.id,
          reason: 'checkpoint_lag',
          lag_ms: lagMs,
        });
      }
    }

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

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

// Backpressure — detect tasks with high backpressure ratio
app.get('/health/flink/backpressure', async (req, res) => {
  try {
    const jobsRes = await axios.get(`${FLINK_API}/jobs`, { timeout: 5000 });
    const runningJobs = jobsRes.data.jobs.filter(j => j.status === 'RUNNING');

    const backpressured = [];

    for (const job of runningJobs) {
      const verticesRes = await axios.get(`${FLINK_API}/jobs/${job.id}`, { timeout: 5000 });
      for (const vertex of verticesRes.data.vertices) {
        const bpRes = await axios.get(
          `${FLINK_API}/jobs/${job.id}/vertices/${vertex.id}/backpressure`,
          { timeout: 5000 }
        );
        const ratio = bpRes.data['backpressure-level'];
        if (ratio === 'high') {
          backpressured.push({ job: job.id, vertex: vertex.name, level: ratio });
        }
      }
    }

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

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

app.listen(3009);

Python Health Sidecar

# health/flink_health.py
import os, time, requests
from flask import Flask, jsonify

app = Flask(__name__)
FLINK_API = os.environ.get("FLINK_API_URL", "http://localhost:8081")
CHECKPOINT_LAG_THRESHOLD = int(os.environ.get("CHECKPOINT_LAG_THRESHOLD_MS", 600000))

@app.route("/health/flink")
def flink_health():
    try:
        r = requests.get(f"{FLINK_API}/overview", timeout=5)
        data = r.json()
        if data.get("taskmanagers", 0) == 0:
            return jsonify({"status": "down", "reason": "no_taskmanagers"}), 503
        if data.get("jobs-failed", 0) > 0:
            return jsonify({"status": "degraded", "jobs_failed": data["jobs-failed"]}), 503
        return jsonify({"status": "ok", "jobs_running": data.get("jobs-running", 0)}), 200
    except Exception as e:
        return jsonify({"status": "down", "error": str(e)}), 503

@app.route("/health/flink/checkpoints")
def checkpoint_health():
    try:
        jobs = [j for j in requests.get(f"{FLINK_API}/jobs", timeout=5).json()["jobs"]
                if j["status"] == "RUNNING"]
        issues = []
        for job in jobs:
            cp = requests.get(f"{FLINK_API}/jobs/{job['id']}/checkpoints", timeout=5).json()
            latest = (cp.get("latest") or {}).get("completed")
            if not latest:
                issues.append({"job": job["id"], "reason": "no_completed_checkpoint"})
                continue
            lag = int(time.time() * 1000) - latest["trigger-timestamp"]
            if lag > CHECKPOINT_LAG_THRESHOLD:
                issues.append({"job": job["id"], "reason": "checkpoint_lag", "lag_ms": lag})
        if issues:
            return jsonify({"status": "degraded", "issues": issues}), 503
        return jsonify({"status": "ok", "jobs_checked": len(jobs)}), 200
    except Exception as e:
        return jsonify({"status": "down", "error": str(e)}), 503

if __name__ == "__main__":
    app.run(port=3009)

Step 2: Configure Vigilmon HTTP Monitors for Flink

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. Set the URL to your Flink health endpoint: https://your-host.example.com/health/flink
  4. Set the check interval to 1 minute
  5. Under Expected response, configure:
    • Status code: 200
    • Response body contains: "status":"ok"
    • Response time threshold: 8000ms
  6. Under Alert channels, assign your on-call Slack channel or PagerDuty integration
  7. Save the monitor

Add monitors for checkpoints and backpressure:

| Monitor | URL | Interval | Priority | |---|---|---|---| | JobManager | /health/flink | 1 minute | P1 | | Checkpoints | /health/flink/checkpoints | 2 minutes | P1 | | Backpressure | /health/flink/backpressure | 5 minutes | P2 |

Vigilmon's multi-region consensus prevents transient checkpoint delays from triggering false positives.

Step 3: Heartbeat Monitoring for Streaming Pipeline SLAs

For streaming pipelines with output SLAs (e.g., "enriched events must reach the sink within 5 minutes of source ingestion"), Vigilmon heartbeat monitors detect when output stops flowing — even when the Flink job shows as RUNNING.

Set Up the Heartbeat Monitor

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Set the name: flink-enrichment-pipeline-output
  3. Set the expected interval: 5 minutes
  4. Set the grace period: 10 minutes
  5. Save — copy the heartbeat URL, e.g. https://vigilmon.online/heartbeat/abc123xyz

Wire Heartbeats Into Your Flink Job (Java)

// VigilmonHeartbeatSink.java
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.URI;
import java.time.Instant;
import java.util.concurrent.atomic.AtomicLong;

public class VigilmonHeartbeatSink<T> extends RichSinkFunction<T> {
    private final String heartbeatUrl;
    private final long intervalMs;
    private transient AtomicLong lastPing;
    private transient HttpClient httpClient;

    public VigilmonHeartbeatSink(String heartbeatUrl, long intervalMs) {
        this.heartbeatUrl = heartbeatUrl;
        this.intervalMs = intervalMs;
    }

    @Override
    public void open(org.apache.flink.configuration.Configuration parameters) {
        this.lastPing = new AtomicLong(0);
        this.httpClient = HttpClient.newHttpClient();
    }

    @Override
    public void invoke(T value, Context context) throws Exception {
        long now = Instant.now().toEpochMilli();
        if (now - lastPing.get() >= intervalMs) {
            HttpRequest req = HttpRequest.newBuilder()
                .uri(URI.create(heartbeatUrl))
                .GET()
                .build();
            httpClient.sendAsync(req, HttpResponse.BodyHandlers.discarding());
            lastPing.set(now);
        }
    }
}

Wire it into your Flink pipeline:

DataStream<EnrichedEvent> enriched = sourceStream
    .keyBy(Event::getCustomerId)
    .process(new EnrichmentFunction())
    .name("enrich");

// Real sink
enriched.sinkTo(kafkaSink).name("kafka-output");

// Vigilmon heartbeat tap — sends ping every 2 minutes of output
enriched.addSink(new VigilmonHeartbeatSink<>(
    System.getenv("VIGILMON_HEARTBEAT_URL"),
    120_000L
)).name("vigilmon-heartbeat").setParallelism(1);

Python (PyFlink)

# vigilmon_sink.py
import os, time, urllib.request
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.functions import SinkFunction

class VigilmonHeartbeatSink(SinkFunction):
    def __init__(self, heartbeat_url: str, interval_seconds: int = 120):
        self._url = heartbeat_url
        self._interval = interval_seconds
        self._last_ping = 0

    def invoke(self, value, context):
        now = time.time()
        if now - self._last_ping >= self._interval:
            try:
                urllib.request.urlopen(self._url, timeout=5)
            except Exception:
                pass
            self._last_ping = now

Step 4: TaskManager Slot Utilization Alerting

When Flink jobs request more slots than are available, new jobs queue or fail to start. Monitor slot pressure:

# Check slot availability via Flink REST API
AVAILABLE=$(curl -s http://localhost:8081/overview | jq '.["slots-available"]')
TOTAL=$(curl -s http://localhost:8081/overview | jq '.["slots-total"]')

if [ "${AVAILABLE}" -lt 2 ]; then
  echo "WARNING: only ${AVAILABLE}/${TOTAL} slots available"
  # Do NOT ping heartbeat — let the slot-pressure monitor alert
else
  curl -fsS "${VIGILMON_SLOT_HEARTBEAT_URL}"
fi

Summary

Flink failures are subtle: jobs stay in RUNNING state while checkpoints lag, backpressure builds, and output stalls. External monitoring with Vigilmon catches all three failure modes before downstream consumers notice:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /health/flink | JobManager availability, failed job count, TaskManager count | | HTTP monitor on /health/flink/checkpoints | Checkpoint completion lag, missing checkpoints | | HTTP monitor on /health/flink/backpressure | High-backpressure task detection | | Heartbeat monitor | Streaming pipeline output SLA — events flowing to sink on schedule |

Get started free at vigilmon.online — your first Flink monitor is running 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