Flink Debug: The Quiet Partition

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?

jobs/sessions.py

# jobs/sessions.py
# Session window aggregator. Reads click events from an 8-partition
# source, sessionizes per user_id with 30-min gap, writes rollups.

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)
    src = src.assign_timestamps_and_watermarks(
        BoundedOutOfOrdernessWatermarks(MAX_OUT_OF_ORDER)
    )
    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

# jobs/sessions.py
# Session window aggregator. Reads click events from an 8-partition
# source, sessionizes per user_id with 30-min gap, writes rollups.

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)
    src = src.assign_timestamps_and_watermarks(
        BoundedOutOfOrdernessWatermarks(MAX_OUT_OF_ORDER)
    )
    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

Active Now|Staff Data Engineer (L6)|||5.9k Attempts|2.3k Solves|