How does Flink handle backpressure

package com.bank.streaming;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class BackpressureDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();

        env.addSource(new KafkaSource<>("nbi-transactions"))
            .keyBy(Transaction::getAccountId)
            .process(new FraudScorer())  // fast
            .addSink(new SlowDatabaseSink()); // SLOW: bottleneck

        // SlowDatabaseSink cannot keep up:
        // 1. Sink's input buffers fill
        // 2. FraudScorer's output buffers fill
        // 3. Source reads fewer Kafka records
        // 4. Kafka lag grows but NO DATA LOST
        env.execute("NBI Backpressure Demo");
    }
}

// No configuration needed!
// Credit-based flow control is ALWAYS active
// It is an emergent property of finite buffers
// You do NOT enable or disable it

package com.bank.streaming;

public class PropagationDemo {
    // Pipeline: Source -> Enricher -> Scorer -> Sink
    //
    // Sink is slow (database writes take 50ms each)
    //
    // Propagation:
    // Step 1: Sink buffers fill (0 credits to Scorer)
    // Step 2: Scorer output buffers fill (0 credits to Enricher)
    // Step 3: Enricher output buffers fill (0 credits to Source)
    // Step 4: Source stops reading from Kafka
    //
    // Kafka lag grows but records stay safely in Kafka
    // When sink catches up, credits flow back and pipeline resumes
}

// Metrics to monitor via Prometheus/Grafana:
// flink_taskmanager_job_task_busyTimeMsPerSecond
// flink_taskmanager_job_task_backPressuredTimeMsPerSecond
// flink_taskmanager_job_task_idleTimeMsPerSecond
// flink_taskmanager_job_task_numRecordsInPerSecond
// flink_taskmanager_job_task_numRecordsOutPerSecond
// kafka_consumer_records_lag_max

package com.bank.streaming;

public class BackpressureFixes {
    // Cause 1: Slow database sink
    // Fix: batch writes, increase sink parallelism, use async writes
    // .addSink(new BatchingDatabaseSink(batchSize=100))
    //     .setParallelism(4);

    // Cause 2: Heavy computation (e.g., ML model scoring)
    // Fix: increase operator parallelism, optimize model
    // .process(new FraudScorer()).setParallelism(8);

    // Cause 3: Data skew on keyBy
    // Fix: salt keys, pre-aggregate, rebalance before keyBy
    // stream.keyBy(txn -> txn.getAccountId() + "-" + txn.hash() % 4)

    // Cause 4: Insufficient parallelism
    // Fix: increase parallelism and add more TaskManagers
    // env.setParallelism(16); // add more slots to cluster
}

Backpressure fix decision tree:

  Is one operator 95%+ busy while upstream is backpressured?
    YES --> that operator is the bottleneck
      Is it a sink?         --> batch writes, more parallelism
      Is it a computation?  --> optimize or increase parallelism
      Is one subtask busy
      while others idle?    --> key skew: salt keys
    NO  --> check for external system issues

package com.bank.streaming;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.environment.CheckpointConfig;

public class UnalignedCheckpointDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(30_000);

        // Unaligned checkpoints: barriers can overtake data in buffers
        // Checkpoints complete faster under backpressure
        env.getCheckpointConfig().enableUnalignedCheckpoints();

        // Trade-off: unaligned checkpoints are larger because
        // they include in-flight buffer data in the snapshot
        env.execute("NBI Unaligned Checkpoint Demo");
    }
}

Aligned vs unaligned checkpoints under backpressure:

  Aligned:   Barrier waits behind data in full buffers
             Checkpoint duration: 30s+ (slow under backpressure)

  Unaligned: Barrier overtakes data in buffers
             In-flight data snapshot with checkpoint
             Checkpoint duration: 2-5s (fast even under backpressure)
             Trade-off: larger checkpoint size

// WRONG: increasing buffer sizes to "fix" backpressure
// This only delays the onset, does not fix the throughput mismatch
// taskmanager.memory.network.fraction: 0.3 (was 0.1)
// Result: buffers fill a bit later, but still fill eventually

// WRONG: dropping records to avoid backpressure
// stream.filter(txn -> shouldDrop(txn)) // data loss!

// CORRECT: fix the bottleneck
// 1. Identify the busy operator
// 2. Increase its parallelism, optimize its logic, or fix its sink
// 3. Backpressure clears naturally when throughput matches

package com.bank.streaming;

public class SeniorBackpressureChecklist {
    // Senior engineer's backpressure checklist:
    //
    // 1. Check Web UI for per-operator backpressure status
    // 2. Find the operator with high busy%, low backpressured%
    //    (this is the bottleneck)
    // 3. Check for subtask skew (one subtask busy, others idle)
    // 4. Check Kafka consumer lag trend
    // 5. Correlate with checkpoint duration
    //    (if checkpoints slow, consider unaligned checkpoints)
    // 6. Fix the bottleneck:
    //    - Slow sink: batch, async, more parallelism
    //    - Heavy compute: optimize, more parallelism
    //    - Key skew: salt keys, pre-aggregate
    // 7. Verify: backpressure clears, lag drains, throughput matches
    //
    // NEVER: increase buffer sizes as a "fix"
    // NEVER: drop records to avoid backpressure
    // NEVER: ignore it (it will get worse)
}