The Quiet Partition

A hard AI Coding interview practice problem on DataDriven.

Domain
AI Coding
Difficulty
hard
Seniority
L6
Asked in
LinkedIn

Problem

Our Flink job aggregates user sessions and writes hourly rollups. Each midnight one source partition (the EU one) falls quiet for ~2 hours during low-traffic. During that window the global watermark stops advancing, no session windows close, and state size grows unbounded until the JM kills the job. Run tests/test_quiet_partition.py to reproduce. Find the watermarking pattern in jobs/sessions.py and fix it. Defend the new behavior: what's the maximum end-to-end latency you've now committed to?

One went quiet. All of them stopped.

Worked solution and explanation

This is a min() over inputs that can go silent, dressed up as a sessions rollup. Flink's global watermark is the minimum of every partition's watermark, so it moves only as fast as the slowest partition. A partition with no events has no new watermark, which means the slowest partition is the one that stopped talking. Anyone can spot that the EU partition goes quiet. What separates candidates is what they do next. Some trust the stale heartbeat comment in jobs/io.py. Others pick an idleness timeout with no number behind it. Do either and the job either deadlocks on a real outage or quietly breaks its 300s latency SLA.

Why one quiet partition freezes everything

Idleness drops a silent partition out of the `min()`, so the other partitions can advance the global watermark on their own.
  • Partition 0 (u_us) (source)
  • Partition 1 (u_eu, quiet 02:00-04:00) (source)
  • Partition 2 (u_apac) (source)
  • Partitions 3-7 (no data) (source)
  • Bounded out-of-orderness + idleness (transform)
  • Global watermark = min(active partitions) (transform)
  • Session window per user_id (30 min gap) (transform)
  • write_sessions (consumer)
  • Partition 0 (u_us) -> Bounded out-of-orderness + idleness
  • Partition 1 (u_eu, quiet 02:00-04:00) -> Bounded out-of-orderness + idleness
  • Partition 2 (u_apac) -> Bounded out-of-orderness + idleness
  • Partitions 3-7 (no data) -> Bounded out-of-orderness + idleness
  • Bounded out-of-orderness + idleness -> Global watermark = min(active partitions)
  • Global watermark = min(active partitions) -> Session window per user_id (30 min gap)
  • Session window per user_id (30 min gap) -> write_sessions

Look again at the source: parallelism=8, but only partitions 0, 1 and 2 ever carry data. In jobs/runner.py, a partition whose watermark is still None empties the candidate list when there is no idleness configured, so global_wm never gets a value. The EU gap is the symptom you were told about. Five partitions that never speak were pinning the watermark from the first event. Without the watermark, no session window fires, per-key state keeps growing, and the JobManager eventually kills the job.

The fix

jobs/sessions.py
from datetime import timedelta
from jobs.flink_lite import StreamEnv, BoundedOutOfOrdernessWatermarks
from jobs.io import read_clicks, write_sessions

GAP = timedelta(minutes=30)
MAX_OUT_OF_ORDER = timedelta(seconds=10)


def build(env: StreamEnv):
    src = env.source(read_clicks, parallelism=8)
    # No heartbeat on this source (constraints.sql), so a silent partition must
    # be declared idle. 60s idle + 10s lag keeps closure well inside the 300s SLA.
    watermarks = BoundedOutOfOrdernessWatermarks(MAX_OUT_OF_ORDER).with_idleness_(timedelta(seconds=60))
    src = src.assign_timestamps_and_watermarks(watermarks)
    sessions = (
        src
        .key_by(lambda e: e['user_id'])
        .session_window(gap=GAP)
        .aggregate(lambda evs: {
            'user_id': evs[0]['user_id'],
            'session_start': min(e['ts'] for e in evs),
            'session_end': max(e['ts'] for e in evs),
            'click_count': len(evs),
        })
    )
    sessions.add_sink(write_sessions)
    return env

One chained call. The duration is written inline so the SLA check can read it.

Step 1: Rule out the heartbeat path

source_partition_behavior says no partition heartbeats, so nothing will advance a silent partition from the outside. That leaves idleness or a wall-clock punctuated assigner. Idleness is the built-in answer and needs no custom code.

Step 2: Chain with_idleness_ onto the strategy

After the timeout, a partition with no events is excluded from the min(). Partitions 3-7 drop out right away. Partition 1 drops out 60s into its quiet window, and it rejoins with its first new event.

Step 3: Size it against job_sla

max_e2e_latency_seconds = 300 is the budget. The worst-case extra delay before a session closes is roughly idleness plus MAX_OUT_OF_ORDER: 60s + 10s = 70s past the gap. That leaves room for checkpoint and sink latency, which the SLA also has to cover.

Idleness too low (say 5s)
A partition that is slow but still alive gets marked idle, the watermark races ahead, and its next events arrive behind global_wm. Windows fire with incomplete counts, and the stragglers become late data.
Idleness too high (say 1h)
The tests pass because the fixture happens to be kind, but a real partition outage holds every session open for up to an hour. That breaks the 300s job_sla long before idleness kicks in, and state keeps growing the whole time.

Common follow-up questions