How do you monitor Flink jobs and what metrics indicate trouble

Flink Web Dashboard tabs:

  Overview:     Job graph with operator parallelism
  Backpressure: Per-operator backpressure status (colored)
  Checkpoints:  History, duration, size, success/failure
  Watermarks:   Event-time progress per operator
  Exceptions:   Recent failures and restarts
  Metrics:      Raw metric values per operator

Access: http://jobmanager-host:8081

package com.bank.monitoring;

// The Web Dashboard is built-in, no code needed
// Access at: http://jobmanager:8081
//
// Key views for NBI fraud pipeline:
// 1. Job graph: verify operator chain and parallelism
// 2. Backpressure tab: spot which operator is bottleneck
// 3. Checkpoints tab: verify checkpoints are completing
// 4. Exceptions tab: check for recent failures

# flink-conf.yaml: enable Prometheus reporter
metrics.reporter.prom.factory.class:
  org.apache.flink.metrics.prometheus.PrometheusReporterFactory
metrics.reporter.prom.port: 9249

# Prometheus scrape config:
scrape_configs:
  - job_name: 'flink'
    static_configs:
      - targets: ['taskmanager-1:9249', 'taskmanager-2:9249']

package com.bank.monitoring;

public class PrometheusMetricsDemo {
    // Key Prometheus queries for Grafana dashboards:
    //
    // Throughput:
    //   flink_taskmanager_job_task_operator_numRecordsInPerSecond
    //   flink_taskmanager_job_task_operator_numRecordsOutPerSecond
    //
    // Checkpoint duration:
    //   flink_jobmanager_job_lastCheckpointDuration
    //
    // Checkpoint size:
    //   flink_jobmanager_job_lastCheckpointFullSize
    //
    // Backpressure:
    //   flink_taskmanager_job_task_busyTimeMsPerSecond
    //   flink_taskmanager_job_task_backPressuredTimeMsPerSecond
    //
    // Kafka lag:
    //   flink_taskmanager_job_task_operator_KafkaConsumer
    //     _records_lag_max
}

Checkpoint duration trend:

  Time     Duration   State Size    Status
  10:00    2.1s       500 MB        Healthy
  10:05    3.4s       650 MB        Growing
  10:10    5.8s       900 MB        Warning
  10:15    12.1s      1.4 GB        Danger
  10:20    TIMEOUT    1.9 GB        STALL

Root cause: state growing without bound
Fix: add state TTL, enable incremental checkpoints,
     increase managed memory

// Prometheus alert for checkpoint duration
// alert: FlinkCheckpointSlowing
// expr: rate(flink_jobmanager_job_lastCheckpointDuration[10m]) > 0.5
// for: 10m
// labels:
//   severity: warning
// annotations:
//   summary: "Checkpoint duration increasing for {{ $labels.job_name }}"

package com.bank.monitoring;

import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;

public class StateLeakFixDemo {
    // PROBLEM: state grows without bound
    // Per-account MapState with entries never removed
    // State size: 500MB -> 1GB -> 2GB -> 4GB -> OOM

    // FIX 1: State TTL (automatic cleanup)
    public static ValueStateDescriptor<String> ttlState() {
        ValueStateDescriptor<String> desc =
            new ValueStateDescriptor<>("accountData", String.class);
        StateTtlConfig ttl = StateTtlConfig
            .newBuilder(Time.hours(24))
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            .cleanupFullSnapshot()
            .build();
        desc.enableTimeToLive(ttl);
        return desc;
    }

    // FIX 2: Explicit cleanup in onTimer
    // Register cleanup timer when state is created
    // Clear state when timer fires
}

// Grafana query to find the bottleneck operator:
// SELECT operator, avg(busyTimeMsPerSecond)
// FROM flink_task_metrics
// WHERE backPressuredTimeMsPerSecond < 50
//   AND busyTimeMsPerSecond > 900
// This finds operators that are busy but NOT backpressured
// = they are the bottleneck

Kafka lag trend:

  Time     Lag (records)   Throughput (rec/s)   Status
  10:00    1,000           10,000               Healthy
  10:05    15,000          8,000                Falling behind
  10:10    50,000          6,000                Growing lag
  10:15    150,000         5,000                Critical

Root cause: throughput < input rate
Fix: find backpressured bottleneck, increase parallelism,
     optimize slow operator, or scale external system

// Prometheus alert for Kafka lag
// alert: FlinkKafkaLagHigh
// expr: flink_kafka_consumer_records_lag_max > 100000
// for: 5m
// labels:
//   severity: critical
// annotations:
//   summary: "Kafka lag exceeding 100K for {{ $labels.topic }}"

Correlation example:

  Observation 1: Kafka lag growing
  Observation 2: Checkpoint duration increasing
  Observation 3: FraudScorer 95% busy, 0% backpressured
  Observation 4: State size growing for FraudScorer

  Root cause story:
    FraudScorer has a state leak (state never cleaned up)
    --> state grows
    --> checkpoint size and duration increase
    --> FraudScorer becomes bottleneck (slow state access)
    --> backpressure propagates upstream
    --> source slows, Kafka lag grows

  Fix: add state TTL to FraudScorer
  Result: state plateaus, checkpoints shrink,
          backpressure clears, lag drains

package com.bank.monitoring;

public class SeniorMonitoringChecklist {
    // Senior monitoring checklist for NBI Flink jobs:
    //
    // 1. Alert on checkpoint duration TREND (rate of change)
    //    not just absolute value
    // 2. Alert on state size GROWTH (sustained increase)
    // 3. Alert on Kafka lag exceeding threshold
    // 4. Monitor restart count (frequent restarts = hidden problem)
    // 5. Monitor failed checkpoint count
    // 6. Correlate: backpressure + lag + busy operator = root cause
    // 7. Latency markers: enable only during investigation
    //    (they add overhead in high-throughput jobs)
    // 8. Set up PagerDuty/OpsGenie alerts for critical metrics
    //
    // Grafana dashboard panels:
    //   Row 1: Throughput (records/sec) per operator
    //   Row 2: Checkpoint duration + size
    //   Row 3: Backpressure per operator (busy/backpressured/idle)
    //   Row 4: Kafka consumer lag
    //   Row 5: State size per operator
    //   Row 6: Restart count + failed checkpoints
}