MCP Tools for Apache Flink — Job health, checkpoint failures, backpressure detection, and TaskManager state

Flink architecture for MCP tool authors

Flink's cluster consists of one JobManager (the coordinator — accepts job submissions, schedules tasks, coordinates checkpoints) and one or more TaskManagers (the workers — execute individual tasks within a job). In session mode (the most common deployment), the JobManager is long-lived and multiple jobs can run concurrently. In application mode, one JobManager exists per job.

The JobManager exposes a REST API (default port 8081) that covers all cluster state. In Kubernetes deployments, this is typically exposed as a ClusterIP service flink-jobmanager:8081. In YARN deployments, the JobManager web URL is printed to the application master logs on startup. MCP tools should discover the JobManager URL dynamically rather than hardcoding it — the URL changes on each YARN application submission.

async function createFlinkClient(jobManagerUrl) {
  const base = jobManagerUrl.replace(/\/$/, '');

  async function get(path) {
    const res = await fetch(`${base}${path}`, {
      headers: { 'Accept': 'application/json' },
      signal: AbortSignal.timeout(10_000),
    });
    if (!res.ok) {
      throw new Error(`Flink API ${path} → ${res.status} ${res.statusText}`);
    }
    return res.json();
  }

  // JobManager liveness — returns cluster info including Flink version
  async function getClusterOverview() {
    return get('/overview');
  }

  // All running, completed, failed jobs
  async function getJobsOverview() {
    return get('/jobs/overview');
  }

  // Full detail for one job including vertex-level task counts
  async function getJobDetail(jobId) {
    return get(`/jobs/${jobId}`);
  }

  // Checkpoint statistics for a running job
  async function getCheckpoints(jobId) {
    return get(`/jobs/${jobId}/checkpoints`);
  }

  // Backpressure ratio for a specific task vertex
  async function getVertexBackpressure(jobId, vertexId) {
    return get(`/jobs/${jobId}/vertices/${vertexId}/backpressure`);
  }

  // TaskManager list and slot availability
  async function getTaskManagers() {
    return get('/taskmanagers');
  }

  return { getClusterOverview, getJobsOverview, getJobDetail, getCheckpoints, getVertexBackpressure, getTaskManagers };
}

Job status and task vertex health

Flink job states follow a state machine: INITIALIZING → CREATED → RUNNING → FAILING → FAILED for error paths, or RUNNING → FINISHED for batch jobs that complete normally. For streaming jobs, FINISHED is an abnormal state — a healthy streaming job stays in RUNNING indefinitely. RESTARTING means the job encountered a task failure and is recovering according to its restart-strategy configuration.

Job-level state: "RUNNING" does not mean all tasks are healthy. Check GET /jobs/{jobId}/vertices — each vertex represents a set of parallel task instances. A vertex in DEPLOYING state means TaskManager slots are being allocated. A vertex in SCHEDULED state means it's waiting for slots — likely a slot shortage (check GET /taskmanagers for freeSlots). A vertex in FAILED state with the job still showing RUNNING means it's within its restart delay.

async function assessJobHealth(flink, jobId) {
  const [job, checkpoints, taskManagers] = await Promise.all([
    flink.getJobDetail(jobId),
    flink.getCheckpoints(jobId),
    flink.getTaskManagers(),
  ]);

  // Job-level state
  const jobState = job.state;  // 'RUNNING' | 'RESTARTING' | 'FAILING' | 'FAILED' | 'FINISHED' | 'CANCELED'
  const isHealthy = jobState === 'RUNNING';
  const isRecovering = jobState === 'RESTARTING';

  // Task-level vertex health
  const vertices = job.vertices ?? [];
  const vertexStates = vertices.map(v => ({
    id: v.id,
    name: v.name,
    parallelism: v.parallelism,
    status: v.status,  // 'RUNNING' | 'DEPLOYING' | 'SCHEDULED' | 'FAILED' | 'CANCELED' | 'FINISHED'
    tasksRunning: v.tasks?.RUNNING ?? 0,
    tasksFailed: v.tasks?.FAILED ?? 0,
    tasksScheduled: v.tasks?.SCHEDULED ?? 0,
    healthy: v.status === 'RUNNING' && (v.tasks?.RUNNING ?? 0) === v.parallelism,
  }));

  const unhealthyVertices = vertexStates.filter(v => !v.healthy);

  // Slot availability
  const totalSlots = taskManagers.taskmanagers?.reduce((sum, tm) => sum + tm.numberOfSlots, 0) ?? 0;
  const freeSlots = taskManagers.taskmanagers?.reduce((sum, tm) => sum + tm.freeSlots, 0) ?? 0;
  const slotsExhausted = freeSlots === 0 && totalSlots > 0;

  // Checkpoint health (streaming jobs only)
  const ckpt = checkpoints;
  const lastCompleted = ckpt?.latest?.completed;
  const checkpointIntervalMs = ckpt?.configuration?.['execution.checkpointing.interval'] ?? 60_000;

  let checkpointHealth = null;
  if (lastCompleted) {
    const ageMs = Date.now() - lastCompleted.trigger_timestamp;
    const durationMs = lastCompleted.end_to_end_duration;
    checkpointHealth = {
      lastCompletedAgeMs: ageMs,
      lastCompletedDurationMs: durationMs,
      failedCount: ckpt.counts?.failed ?? 0,
      completedCount: ckpt.counts?.completed ?? 0,
      // Stale if last checkpoint is older than 3x the configured interval
      stale: ageMs > checkpointIntervalMs * 3,
      // Slow if duration exceeds 2x the interval (checkpoint backpressuring the job)
      slow: durationMs > checkpointIntervalMs * 2,
    };
  }

  return {
    jobId,
    jobName: job.name,
    jobState,
    isHealthy,
    isRecovering,
    startTime: job['start-time'],
    duration: job.duration,
    vertices: vertexStates,
    unhealthyVertices,
    slotAvailability: { totalSlots, freeSlots, slotsExhausted },
    checkpointHealth,
    // Overall health: job running + all vertices healthy + checkpoints not stale
    overallHealthy: isHealthy && unhealthyVertices.length === 0 && (!checkpointHealth || (!checkpointHealth.stale && !checkpointHealth.slow)),
  };
}

Checkpoint health: the most important Flink signal

Checkpointing is Flink's mechanism for fault tolerance and exactly-once semantics. At each checkpoint interval, Flink snapshots the state of all operators and writes it to a state backend (filesystem, RocksDB, S3). Checkpoint failures mean the job cannot recover to a consistent state after a task failure — a job with frequent checkpoint failures is running without a safety net.

The checkpoint endpoint GET /jobs/{jobId}/checkpoints returns a counts object with completed, failed, in_progress, and restored counts. An increasing failed count across successive polls is the primary alert signal — individual failures are normal (network hiccup during snapshot), but sustained failures indicate state backend problems (S3 throttling, RocksDB compaction lag, checkpoint size exceeding timeout).

The latest.completed object gives the most recent successful checkpoint's end_to_end_duration and state_size. Checkpoint duration growing over time signals state growth (RocksDB LSM compaction not keeping up) or output backpressure (checkpointing is blocked by downstream slow consumers). When end_to_end_duration exceeds the execution.checkpointing.timeout (default 10 minutes), the checkpoint is cancelled and counted as failed.

async function checkpointHealthCheck(flink, jobId, checkpointIntervalMs = 60_000) {
  const ckpt = await flink.getCheckpoints(jobId);

  if (!ckpt || !ckpt.counts) {
    return { healthy: false, reason: 'checkpoint_api_unavailable' };
  }

  const { counts, latest, history, configuration } = ckpt;
  const configuredInterval = configuration?.['execution.checkpointing.interval'] ?? checkpointIntervalMs;
  const configuredTimeout = configuration?.['execution.checkpointing.timeout'] ?? 600_000;

  // Last completed checkpoint freshness
  const lastCompleted = latest?.completed;
  const lastFailed = latest?.failed;

  const now = Date.now();
  const completedAgeMs = lastCompleted ? now - lastCompleted.trigger_timestamp : null;
  const stale = completedAgeMs !== null && completedAgeMs > configuredInterval * 3;

  // Checkpoint failure rate over recent history
  const recentHistory = (history ?? []).slice(0, 10);  // Last 10 checkpoints
  const recentFailed = recentHistory.filter(c => c.status === 'FAILED').length;
  const failureRate = recentHistory.length > 0 ? recentFailed / recentHistory.length : 0;

  // Current in-progress checkpoint duration (should not exceed timeout)
  const inProgress = latest?.in_progress;
  const inProgressDurationMs = inProgress ? now - inProgress.trigger_timestamp : null;
  const checkpointStuck = inProgressDurationMs !== null && inProgressDurationMs > configuredTimeout * 0.8;

  const alerts = [];
  if (stale) alerts.push(`Last completed checkpoint is ${Math.round(completedAgeMs / 1000)}s old (threshold: ${Math.round(configuredInterval * 3 / 1000)}s)`);
  if (failureRate > 0.3) alerts.push(`${Math.round(failureRate * 100)}% of recent checkpoints failed`);
  if (checkpointStuck) alerts.push(`In-progress checkpoint running for ${Math.round(inProgressDurationMs / 1000)}s — approaching timeout`);
  if (lastCompleted && lastCompleted.end_to_end_duration > configuredInterval * 2) {
    alerts.push(`Last checkpoint took ${Math.round(lastCompleted.end_to_end_duration / 1000)}s — exceeds 2x interval`);
  }

  return {
    healthy: alerts.length === 0,
    alerts,
    counts,
    lastCompleted: lastCompleted ? {
      checkpointId: lastCompleted.id,
      triggerTimestamp: lastCompleted.trigger_timestamp,
      durationMs: lastCompleted.end_to_end_duration,
      stateSizeBytes: lastCompleted.state_size,
      completedAgeMs,
    } : null,
    failureRate,
    inProgress: inProgress ? {
      checkpointId: inProgress.id,
      runningForMs: inProgressDurationMs,
      stuckAlert: checkpointStuck,
    } : null,
  };
}

Backpressure: when RUNNING means stalled

Backpressure is the most critical subtle health signal in Flink. A job in RUNNING state with all vertices RUNNING can still be effectively stalled if backpressure is propagating upstream from a slow sink. Backpressure occurs when a downstream operator cannot consume records as fast as an upstream operator produces them — the upstream operator blocks waiting for buffer space, consuming CPU on waiting rather than processing.

The endpoint GET /jobs/{jobId}/vertices/{vertexId}/backpressure returns a backpressure-level of OK, LOW, HIGH, or BLOCKING, and a backpressure ratio from 0.0 to 1.0 (the fraction of time the vertex spent waiting for output buffer space). The status field is deprecated in newer Flink versions — use backpressure-level directly.

async function assessBackpressure(flink, jobId) {
  const job = await flink.getJobDetail(jobId);
  const vertices = job.vertices ?? [];

  // Probe backpressure for all vertices in parallel (one request per vertex)
  const backpressureResults = await Promise.allSettled(
    vertices.map(async (vertex) => {
      const bp = await flink.getVertexBackpressure(jobId, vertex.id);
      return {
        vertexId: vertex.id,
        vertexName: vertex.name,
        parallelism: vertex.parallelism,
        backpressureLevel: bp['backpressure-level'],  // 'OK' | 'LOW' | 'HIGH' | 'BLOCKING'
        backpressureRatio: bp.backpressure,            // 0.0 to 1.0
        idleRatio: bp.idleness,                        // 0.0 to 1.0 — time waiting for input (source starved)
        busyRatio: bp.busiest,                         // 0.0 to 1.0 — time actively processing
        subtasks: bp.subtasks ?? [],                   // Per-parallel-instance detail
      };
    })
  );

  const vertexBP = backpressureResults
    .filter(r => r.status === 'fulfilled')
    .map(r => r.value);

  // Find the bottleneck: the vertex with the highest backpressure that has LOW idle (it's the source of pressure)
  const highBackpressure = vertexBP.filter(v => v.backpressureLevel === 'HIGH' || v.backpressureLevel === 'BLOCKING');
  const bottleneck = highBackpressure.sort((a, b) => (b.backpressureRatio ?? 0) - (a.backpressureRatio ?? 0))[0] ?? null;

  // Source starvation: sources with high idleness are waiting for input (different problem — upstream lag)
  const starvedSources = vertexBP.filter(v => (v.idleRatio ?? 0) > 0.8 && v.vertexName.toLowerCase().includes('source'));

  return {
    jobId,
    hasHighBackpressure: highBackpressure.length > 0,
    bottleneck: bottleneck ? {
      name: bottleneck.vertexName,
      ratio: bottleneck.backpressureRatio,
      level: bottleneck.backpressureLevel,
    } : null,
    starvedSources: starvedSources.map(s => ({ name: s.vertexName, idleRatio: s.idleRatio })),
    allVertices: vertexBP,
  };
}

Watermark lag and event-time processing health

For event-time streaming jobs, watermark lag measures how far behind the job's processing time is from real-time. Watermarks advance the job's notion of "current time" — when watermarks stop advancing, windows stop closing, aggregations stall, and the job appears to produce no output even while RUNNING. Watermark lag is distinct from backpressure: a job can have zero backpressure but growing watermark lag if event timestamps in the source data are old.

The Flink REST API exposes watermarks per source vertex via GET /jobs/{jobId}/vertices/{sourceVertexId}/subtasks/metrics?get=currentOutputWatermark. The watermark value is Unix time in milliseconds. Lag = System.currentTimeMillis() - min(watermark across all source subtasks). Alert when lag exceeds your maximum acceptable latency (typically 2–5× your window size).

async function watermarkLagCheck(flink, jobId) {
  const job = await flink.getJobDetail(jobId);

  // Find source vertices (vertices with no upstream edges)
  const sourceVertices = (job.vertices ?? []).filter(v =>
    v.inputs === null || v.inputs.length === 0
  );

  const watermarks = await Promise.allSettled(
    sourceVertices.map(async (vertex) => {
      // Query watermark metric across all subtasks
      const metricRes = await fetch(
        `${flink.baseUrl}/jobs/${jobId}/vertices/${vertex.id}/subtasks/metrics?get=currentOutputWatermark`,
        { headers: { 'Accept': 'application/json' } }
      ).then(r => r.json());

      // currentOutputWatermark is in epoch milliseconds; Long.MIN_VALUE (-9223372036854775808) means no watermark
      const MIN_WATERMARK = -9223372036854775808;
      const values = (metricRes ?? [])
        .map(m => parseFloat(m.value))
        .filter(v => v > MIN_WATERMARK && !isNaN(v));

      const minWatermark = values.length > 0 ? Math.min(...values) : null;
      const lagMs = minWatermark !== null ? Date.now() - minWatermark : null;

      return {
        vertexId: vertex.id,
        vertexName: vertex.name,
        minWatermarkMs: minWatermark,
        lagMs,
        lagSeconds: lagMs !== null ? Math.round(lagMs / 1000) : null,
        hasWatermark: minWatermark !== null,
      };
    })
  );

  const results = watermarks
    .filter(r => r.status === 'fulfilled')
    .map(r => r.value)
    .filter(r => r.hasWatermark);

  const maxLagMs = results.length > 0 ? Math.max(...results.map(r => r.lagMs ?? 0)) : null;

  return {
    sources: results,
    maxLagMs,
    maxLagSeconds: maxLagMs !== null ? Math.round(maxLagMs / 1000) : null,
    // Alert if any source has lag > 5 minutes (tune to your window size)
    alert: maxLagMs !== null && maxLagMs > 5 * 60 * 1000,
  };
}

AliveMCP integration for Flink cluster monitoring

Flink health monitoring requires three separate probe layers. Register each as an independent AliveMCP check with different alert thresholds. The JobManager availability probe runs every 60 seconds with a 5-second timeout — if the JobManager is unreachable, all streaming jobs are dead. The job health probe runs every 60 seconds with a 15-second timeout (it fans out to multiple endpoints). The checkpoint health probe runs every 5 minutes with a 10-second timeout.

async function flinkClusterHealthProbe(jobManagerUrl, targetJobIds = []) {
  const flink = await createFlinkClient(jobManagerUrl);

  // Layer 1: JobManager availability
  let clusterOverview;
  try {
    clusterOverview = await flink.getClusterOverview();
  } catch (err) {
    return { healthy: false, layer: 'jobmanager', error: err.message };
  }

  // Layer 2: Job health (all specified jobs, or all running jobs)
  const jobsOverview = await flink.getJobsOverview();
  const runningJobs = (jobsOverview.jobs ?? []).filter(j => j.state === 'RUNNING');
  const jobsToCheck = targetJobIds.length > 0
    ? runningJobs.filter(j => targetJobIds.includes(j.jid))
    : runningJobs;

  const jobHealthResults = await Promise.allSettled(
    jobsToCheck.map(j => assessJobHealth(flink, j.jid))
  );

  const jobHealthSummary = jobHealthResults
    .filter(r => r.status === 'fulfilled')
    .map(r => r.value);

  const failingJobs = jobHealthSummary.filter(j => !j.overallHealthy);

  // Layer 3: TaskManager slots
  const taskManagers = await flink.getTaskManagers();
  const totalSlots = taskManagers.taskmanagers?.reduce((s, tm) => s + tm.numberOfSlots, 0) ?? 0;
  const freeSlots = taskManagers.taskmanagers?.reduce((s, tm) => s + tm.freeSlots, 0) ?? 0;

  return {
    healthy: failingJobs.length === 0,
    cluster: {
      flinkVersion: clusterOverview['flink-version'],
      slotsTotal: clusterOverview['slots-total'],
      slotsAvailable: clusterOverview['slots-available'],
      taskManagers: clusterOverview['taskmanagers'],
      jobsRunning: clusterOverview['jobs-running'],
    },
    jobs: jobHealthSummary,
    failingJobs: failingJobs.map(j => ({
      jobId: j.jobId,
      name: j.jobName,
      state: j.jobState,
      unhealthyVertices: j.unhealthyVertices,
      checkpointHealth: j.checkpointHealth,
    })),
    slots: { total: totalSlots, free: freeSlots, exhausted: freeSlots === 0 && totalSlots > 0 },
  };
}

Set AliveMCP alert thresholds: critical when JobManager is unreachable or any streaming job enters FAILED state; warning when checkpoint failure rate exceeds 30%, last completed checkpoint is older than 3× the interval, or backpressure ratio exceeds 0.8 on any vertex for more than two consecutive probe cycles.

Related guides

• MCP tools for Apache Spark — SparkContext health, stage progress, executor state, and shuffle diagnostics
• MCP tools for Apache Kafka — consumer group lag, partition health, and broker availability
• MCP tools for ClickHouse — ReplicatedMergeTree health, system.parts, and replication queue monitoring
• MCP tools for Trino — coordinator health, worker state, query execution stages, and memory pool management
• MCP server health check patterns — composite endpoints and failure classification
• MCP server uptime monitoring — probes, intervals, and alert routing