Backpressure

Backpressure scenario:

Source (100K events/sec) → Map → KeyBy → Window → Sink (50K events/sec)
                                                         ↑
                                                    BOTTLENECK!

Without flow control:
  Source produces 100K/sec
  Sink can only handle 50K/sec
  Buffer grows by 50K events/sec
  After 60 seconds: 3M events buffered
  After 5 minutes: OOM crash 💥

With Flink's credit-based flow control:
  Sink signals "I'm full" to Window operator
  Window signals "I'm full" to KeyBy
  KeyBy signals "I'm full" to Map
  Map signals "I'm full" to Source
  Source slows down to 50K/sec (matches sink capacity)
  → No unbounded buffering, no OOM
  → But: end-to-end latency increases
  → And: Kafka consumer lag grows (source reads slower)

Credit-Based Flow Control:

Sender (upstream)                    Receiver (downstream)
┌──────────────┐                    ┌──────────────┐
│ Output Buffer│ ──── data ────→    │ Input Buffer │
│              │ ←── credits ───    │              │
└──────────────┘                    └──────────────┘

1. Receiver has N input buffers available (e.g., 4)
2. Receiver sends credits=4 to sender
3. Sender sends up to 4 buffers of data
4. Receiver processes buffers, frees them
5. Receiver sends new credits as buffers are freed
6. If receiver is slow → no credits sent → sender blocks

Network buffer pool per TaskManager:
  - Configured via taskmanager.memory.network.fraction (default 10%)
  - Split into input buffers and output buffers
  - Each channel (connection between operators) gets exclusive buffers
  - Floating buffers shared across channels for flexibility

Backpressure propagation:
  Sink full → no credits to Window → Window blocks
  Window full → no credits to KeyBy → KeyBy blocks
  KeyBy full → no credits to Source → Source stops reading Kafka
  → Kafka consumer lag increases (visible externally)

Detection Methods:

1. Flink Web UI — Backpressure Tab
   Each operator shows: OK / LOW / HIGH
   - OK: < 10% of time backpressured
   - LOW: 10-50% of time backpressured
   - HIGH: > 50% of time backpressured

   Reading the UI:
   Source [OK] → Map [OK] → Window [HIGH] → Sink [OK]
                                    ↑
   The FIRST operator showing HIGH is the bottleneck's UPSTREAM
   The actual bottleneck is the operator AFTER the backpressured one
   → Sink is the bottleneck (Window is backpressured because Sink is slow)

2. Metrics (Prometheus/Grafana)
   - outPoolUsage: % of output buffers in use (high = sending blocked)
   - inPoolUsage: % of input buffers in use (high = receiving backed up)
   - numRecordsOutPerSecond: throughput (dropping = backpressure)
   - busyTimeMsPerSecond: time operator is busy (1000 = fully saturated)

3. Checkpoint Duration
   - Long checkpoint duration often indicates backpressure
   - Barrier alignment takes longer when buffers are full
   - Check "Alignment Duration" in checkpoint details

4. Kafka Consumer Lag
   - Growing lag = source is reading slower than production rate
   - Often caused by downstream backpressure slowing the source

// Strategy 1: Increase parallelism of bottleneck operator
stream
    .keyBy(Event::getUserId)
    .process(new ExpensiveFunction())
    .setParallelism(32)  // increase just this operator
    .name("Expensive Processing");

// Strategy 2: Async I/O for slow external calls
AsyncDataStream.unorderedWait(
    stream,
    new AsyncDatabaseLookup(),
    30, TimeUnit.SECONDS,
    200  // 200 concurrent requests (up from default)
);

// Strategy 3: Pre-aggregate to reduce downstream load
stream
    .keyBy(Event::getRegion)
    .window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
    .reduce((a, b) -> merge(a, b))  // pre-aggregate
    .keyBy(Result::getRegion)
    .process(new ExpensiveSink());  // less data to process

// Strategy 4: Handle data skew with salting
stream
    .map(event -> {
        // Add random salt to hot keys
        String saltedKey = event.getKey() + "_" + random.nextInt(10);
        return event.withKey(saltedKey);
    })
    .keyBy(Event::getKey)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .reduce(reducer)
    // Second stage: aggregate salt partitions
    .keyBy(result -> result.getOriginalKey())
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .reduce(reducer);

Relationship between backpressure and Kafka lag:

Normal operation (no backpressure):
  Kafka production rate: 100K events/sec
  Flink consumption rate: 100K events/sec
  Consumer lag: ~0 (caught up)
  Backpressure: none

Backpressure scenario:
  Kafka production rate: 100K events/sec
  Flink processing capacity: 60K events/sec (bottleneck downstream)
  Flink consumption rate: 60K events/sec (throttled by backpressure)
  Consumer lag: growing by 40K events/sec
  Backpressure: HIGH on operators upstream of bottleneck

Key insight:
  Backpressure → Source reads slower → Lag grows
  Lag is the SYMPTOM, backpressure is the MECHANISM

Lag without backpressure (different problem):
  - Source parallelism < Kafka partitions (some partitions unread)
  - Deserialization is the bottleneck (source itself is slow)
  - Network issues between Flink and Kafka

Monitoring both:
  - Kafka lag: tells you HOW FAR BEHIND you are
  - Backpressure metrics: tell you WHERE the bottleneck is
  - Together: diagnose both the severity and the cause