tutorial

How to Monitor Apache Flink SQL with Vigilmon

"A step-by-step guide to monitoring Apache Flink SQL jobs—covering JobManager health, TaskManager slots, checkpoint duration, watermark lag, and Vigilmon integration."

How to Monitor Apache Flink SQL with Vigilmon

Apache Flink is the de facto standard for stateful stream processing at scale. Flink SQL brings that power to SQL practitioners, letting teams run continuous queries over Kafka topics, files, and other sources without writing Java or Python. But running Flink SQL in production means owning its operational complexity: JobManagers can fail, TaskManager slots can exhaust, checkpoint durations can spike, and watermarks can lag so far behind real time that event-time queries return wrong results.

This tutorial walks through monitoring Flink SQL end-to-end—from JobManager health checks to per-job watermark lag—and wiring everything into Vigilmon for automated alerting.

Prerequisites

  • Apache Flink 1.17+ cluster (standalone, YARN, or Kubernetes)
  • At least one Flink SQL job running
  • Vigilmon account at vigilmon.online
  • curl and jq for JSON parsing

Flink's Architecture

Flink clusters have two main components:

  • JobManager: The control plane. It accepts job submissions, coordinates checkpoints, and manages failover. There is typically one active JobManager (or a standby in HA mode).
  • TaskManagers: The workers. Each TaskManager provides a fixed number of task slots—the unit of parallelism. Each slot runs one parallel subtask of a Flink job.

Flink SQL jobs are translated into Flink DataStream jobs internally. The same monitoring approach applies to both.

Step 1: Verify JobManager Health

Flink exposes a REST API from the JobManager on port 8081 by default. The most basic health check confirms the REST API is responding:

# Check JobManager REST API
curl -sf http://flink-jobmanager:8081/overview

A successful response looks like:

{
  "taskmanagers": 4,
  "slots-total": 32,
  "slots-available": 12,
  "jobs-running": 3,
  "jobs-finished": 12,
  "jobs-cancelled": 0,
  "jobs-failed": 1,
  "flink-version": "1.18.0"
}

Parse this to extract the fields you care about:

#!/bin/bash
FLINK_URL="http://flink-jobmanager:8081"

overview=$(curl -sf "$FLINK_URL/overview")
if [ $? -ne 0 ]; then
  echo "CRITICAL: Cannot reach Flink JobManager"
  exit 2
fi

failed=$(echo "$overview" | jq -r '.["jobs-failed"]')
running=$(echo "$overview" | jq -r '.["jobs-running"]')
tms=$(echo "$overview" | jq -r '.taskmanagers')

echo "JobManager OK: $running jobs running, $failed failed, $tms TaskManagers"

if [ "$failed" -gt "0" ]; then
  echo "WARNING: $failed failed jobs detected"
  exit 1
fi

For HA deployments, check both the active JobManager and its standby. If you use ZooKeeper or Kubernetes for HA, verify the leader election is healthy as well.

Step 2: Monitor TaskManager Slot Availability

Slots are the scarce resource in Flink. When all slots are consumed, new jobs cannot start and failed subtasks cannot recover. The /overview endpoint gives you aggregate slot counts. For per-TaskManager detail:

# List all TaskManagers and their slot usage
curl -sf http://flink-jobmanager:8081/taskmanagers | jq '
  .taskmanagers[] | {
    id: .id,
    host: .path,
    slots_total: .["hardware"]["cpu-cores"],
    free_slots: .["free-slots"],
    mem_total_mb: (.["hardware"]["memory-managed"] / 1048576 | floor)
  }
'

Write a slot availability check:

#!/bin/bash
FLINK_URL="http://flink-jobmanager:8081"
MIN_FREE_SLOTS=2  # alert if fewer than 2 slots available cluster-wide

overview=$(curl -sf "$FLINK_URL/overview")
free_slots=$(echo "$overview" | jq -r '.["slots-available"]')
total_slots=$(echo "$overview" | jq -r '.["slots-total"]')

echo "Slots: $free_slots / $total_slots available"

if [ "$free_slots" -lt "$MIN_FREE_SLOTS" ]; then
  echo "WARNING: Only $free_slots free slots remain (minimum $MIN_FREE_SLOTS required)"
  exit 1
fi
echo "OK: Slot availability healthy"

When you see free slots consistently near zero, it is time to add TaskManagers or reduce job parallelism.

Step 3: Monitor SQL Job Checkpoint Duration

Checkpoints are Flink's mechanism for fault tolerance. When a job fails, Flink restarts from the last successful checkpoint. If checkpoints take too long, they can overlap and eventually fail, leaving your job in a half-functioning state where it continuously restarts.

List all running jobs and their checkpoint statistics:

# Get all running jobs
jobs=$(curl -sf http://flink-jobmanager:8081/jobs | jq -r '.jobs[] | select(.status == "RUNNING") | .id')

for job_id in $jobs; do
  echo "=== Job: $job_id ==="
  curl -sf "http://flink-jobmanager:8081/jobs/$job_id/checkpoints" | jq '
    .latest | {
      completed: .completed.id,
      duration_ms: .completed.duration,
      size_bytes: .completed.state_size,
      alignment_buffered: .completed["alignment-buffered"],
      failed_since_last: .counts.failed
    }
  '
done

Build a checkpoint health check with thresholds:

#!/bin/bash
FLINK_URL="http://flink-jobmanager:8081"
MAX_CHECKPOINT_MS=60000  # 60 seconds
exit_code=0

jobs=$(curl -sf "$FLINK_URL/jobs" | jq -r '.jobs[] | select(.status == "RUNNING") | .id')

for job_id in $jobs; do
  checkpoints=$(curl -sf "$FLINK_URL/jobs/$job_id/checkpoints")
  duration=$(echo "$checkpoints" | jq -r '.latest.completed.duration // 0')
  failed=$(echo "$checkpoints" | jq -r '.counts.failed // 0')

  if [ "$failed" -gt "0" ]; then
    echo "CRITICAL: Job $job_id has $failed failed checkpoints"
    exit_code=2
  fi

  if [ "$duration" -gt "$MAX_CHECKPOINT_MS" ]; then
    echo "WARNING: Job $job_id checkpoint took ${duration}ms (max ${MAX_CHECKPOINT_MS}ms)"
    exit_code=$(( exit_code > 1 ? exit_code : 1 ))
  else
    echo "OK: Job $job_id checkpoint in ${duration}ms"
  fi
done

exit $exit_code

Step 4: Track Watermark Lag

Watermarks tell Flink how far behind real time the event stream is. If a watermark falls far behind, event-time windows stop triggering and your SQL queries over time windows return no results or silently accumulate records without emitting.

Flink exposes per-operator watermarks through the Prometheus metrics endpoint (port 9249 by default, or via Flink's metrics reporter). The key metric is currentOutputWatermark, which you compare against wall-clock time:

# Pull watermark metrics (if Prometheus reporter is enabled)
curl -s http://flink-taskmanager:9249/metrics | grep 'currentOutputWatermark'

If you are using the built-in metrics reporter, you can also query watermarks via the REST API for individual job vertices:

#!/bin/bash
FLINK_URL="http://flink-jobmanager:8081"
MAX_WATERMARK_LAG_SECONDS=30

# Get current epoch in milliseconds
NOW_MS=$(date +%s%3N)

jobs=$(curl -sf "$FLINK_URL/jobs" | jq -r '.jobs[] | select(.status == "RUNNING") | .id')

for job_id in $jobs; do
  vertices=$(curl -sf "$FLINK_URL/jobs/$job_id" | jq -r '.vertices[].id')
  for vertex_id in $vertices; do
    metrics=$(curl -sf "$FLINK_URL/jobs/$job_id/vertices/$vertex_id/metrics?get=0.currentOutputWatermark")
    watermark=$(echo "$metrics" | jq -r '.[0].value // "-9223372036854775808"')

    if [ "$watermark" = "-9223372036854775808" ]; then
      echo "INFO: Job $job_id vertex $vertex_id has no watermark yet"
      continue
    fi

    lag_ms=$(( NOW_MS - watermark ))
    lag_s=$(( lag_ms / 1000 ))

    if [ "$lag_s" -gt "$MAX_WATERMARK_LAG_SECONDS" ]; then
      echo "WARNING: Job $job_id vertex $vertex_id watermark lag ${lag_s}s"
    else
      echo "OK: Job $job_id vertex $vertex_id watermark lag ${lag_s}s"
    fi
  done
done

Step 5: Monitor State Backend Size

Flink jobs accumulate keyed state—tables in RocksDB or heap memory—that grows with the number of distinct keys and the time-to-live of your state. An unbounded state backend eventually causes out-of-memory failures or checkpoint timeouts.

Flink exposes state size per operator in checkpoint summaries:

# Check state size in the latest checkpoint
curl -sf "http://flink-jobmanager:8081/jobs/$JOB_ID/checkpoints" | jq '
  .latest.completed | {
    checkpoint_id: .id,
    total_state_bytes: .state_size,
    duration_ms: .duration
  }
'

For RocksDB state backends, also watch the JVM heap and native memory through the TaskManager metrics:

# TaskManager JVM heap usage
curl -sf "http://flink-jobmanager:8081/taskmanagers/$TM_ID/metrics?get=Status.JVM.Memory.Heap.Used,Status.JVM.Memory.Heap.Max" \
  | jq '.[] | {metric: .id, value: .value}'

Set an alert when total state size per job exceeds 80% of the checkpoint storage budget.

Step 6: Configure Vigilmon Endpoint Checks

JobManager Health Check

In Vigilmon, navigate to Monitors → New Monitor:

| Field | Value | |-------|-------| | Name | Flink JobManager API | | Type | HTTP | | URL | http://flink-jobmanager:8081/overview | | Expected Status | 200 | | Response Contains | "jobs-running" | | Interval | 30 seconds | | Timeout | 10 seconds |

Job Failure Check

Create a second monitor pointing to your jobs endpoint:

| Field | Value | |-------|-------| | Name | Flink Running Jobs | | Type | HTTP | | URL | http://flink-jobmanager:8081/jobs | | Expected Status | 200 | | Interval | 60 seconds |

Use Vigilmon's response validation to check that no job has status FAILED. Alternatively, wrap the check in a lightweight API:

#!/usr/bin/env python3
from flask import Flask, jsonify
import requests

app = Flask(__name__)
FLINK_URL = "http://flink-jobmanager:8081"

@app.route('/flink/health')
def flink_health():
    try:
        overview = requests.get(f"{FLINK_URL}/overview", timeout=5).json()
        jobs = requests.get(f"{FLINK_URL}/jobs", timeout=5).json()

        failed = [j for j in jobs.get('jobs', []) if j['status'] == 'FAILED']
        running = [j for j in jobs.get('jobs', []) if j['status'] == 'RUNNING']

        ok = len(failed) == 0 and overview.get('taskmanagers', 0) > 0
        return jsonify({
            'status': 'ok' if ok else 'degraded',
            'running_jobs': len(running),
            'failed_jobs': len(failed),
            'taskmanagers': overview.get('taskmanagers', 0),
            'free_slots': overview.get('slots-available', 0),
        }), 200 if ok else 503

    except Exception as e:
        return jsonify({'status': 'error', 'error': str(e)}), 503

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

Point Vigilmon at http://your-api:8080/flink/health expecting status 200 and response containing "status": "ok".

Checkpoint and Watermark Checks

For checkpoint and watermark monitoring, use Vigilmon's Script monitor type with the bash scripts from Steps 3 and 4. Schedule them every 5 minutes—checkpoint issues typically manifest over several minutes, not seconds.

Alert Routing

Under Settings → Notifications in Vigilmon:

  • JobManager down: PagerDuty immediately (blocking all jobs)
  • Failed jobs: Slack #flink-ops + email in 5 minutes
  • Checkpoint failures: Slack #flink-ops immediately
  • Watermark lag: Slack #data-quality after 2 consecutive failures
  • Slot exhaustion: Email within 10 minutes

Flink SQL-Specific Checks

When using Flink's SQL client or Table API, add these checks specific to SQL workloads:

# Check Flink SQL Gateway (if using Flink SQL Gateway)
curl -sf http://flink-sql-gateway:8083/v1/info

# List active sessions
curl -sf http://flink-sql-gateway:8083/v1/sessions

# List operations in a session
curl -sf "http://flink-sql-gateway:8083/v1/sessions/$SESSION_ID/operations"

If a SQL Gateway session accumulates operations without completing them, it is leaking resources. Alert if any session has more than 50 pending operations.

Monitoring Checklist

| Component | Check | Threshold | |-----------|-------|-----------| | JobManager | HTTP /overview | 200 OK | | TaskManagers | Registered TM count | ≥ 1 | | Slots | Free slot count | ≥ MIN_FREE_SLOTS | | Jobs | FAILED job count | 0 | | Checkpoints | Duration | < 60 seconds | | Checkpoints | Failure count | 0 | | Watermarks | Lag behind wall clock | < 30 seconds | | State backend | Total state size | < 80% of budget | | SQL Gateway | HTTP /v1/info | 200 OK |

Troubleshooting

JobManager unreachable:

  • Check pod/process status: kubectl get pods -l component=jobmanager -n flink
  • Check ZooKeeper leadership if running HA
  • Review GC logs—long GC pauses prevent heartbeats

Slots exhausted:

  • Identify which jobs are consuming slots: check running job parallelism
  • Add TaskManager replicas or increase slot count per TaskManager
  • Cancel stuck jobs that are not processing data

Checkpoint timeouts:

  • Look for slow operators in the checkpoint alignment phase
  • Reduce the number of keys in state to limit state size
  • Switch from in-memory state backend to RocksDB for large state

Watermark stuck:

  • Check the source operator for idle partitions—one idle Kafka partition can block the entire watermark
  • Enable allowedLateness and idle source timeout in your Flink SQL source connector
  • Verify event timestamps in the source topic are monotonically increasing

Conclusion

Flink SQL is a powerful way to run continuous queries over streams, but it inherits Flink's operational complexity. JobManagers fail, slots fill up, checkpoints slow down, and watermarks stall—often silently. By combining Flink's REST API with Vigilmon's monitoring infrastructure, you get automated health checks that catch these issues before they cause data quality problems or outages.

Start with the JobManager health check and failed job detection in Vigilmon, then layer in checkpoint duration alerts and watermark lag monitoring. Once those are stable, add state size tracking to prevent the gradual resource exhaustion that kills long-running Flink jobs in production.

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