Late events в деталях

В уроке 05.05 мы уже разбирали базовое использование allowedLateness и sideOutputLateData. Но в production-сценариях late events — это не просто конфигурационный параметр, а целая стратегия: как мониторить, как реагировать, как балансировать корректность и стоимость.

В этом уроке углубляемся: типичные источники поздних событий, паттерны их обработки, антипаттерны и production-чеклист.

Откуда берутся late events

Чтобы правильно настроить пайплайн, нужно понимать причины:

1. Network latency между регионами. События из удалённого DC приходят с дополнительной задержкой 100-500ms. На фоне 5-секундного boundedOutOfOrderness — это late.

2. Kafka producer batching. Producer с linger.ms=100 накапливает сообщения 100ms перед отправкой. Если в этот момент происходит restart broker-а — пачка приходит с задержкой в секунды.

3. Backfill через replay. Re-process исторических данных через Kafka offsets: события 30-дневной давности приходят сейчас. Это extremely late events.

4. Mobile/IoT clients. Устройство офлайн часами или сутками, потом отправляет накопленные события. Может быть несколько часов опоздания.

5. Consumer rebalance. При переключении partition между consumer-инстансами есть короткий период (секунды), когда сообщения “копятся” — потом приходят пачкой.

6. Source/sink failures. Retry-логика producer-ов добавляет задержку. У части пакетов задержка может быть кратно дольше нормальной.

7. Time clock drift. Если producer ставит timestamp по своим часам, а они отстают от server-clock — все события производителя выглядят late.

Понимание источника — ключ к правильному выбору allowedLateness. Для случая (1-2) обычно достаточно 1-5 минут. Для (4) может требоваться часы. Для (3) лучше использовать withWatermarkAlignment (см. урок 04).

Multiple firings: окно эмиттит несколько раз

С allowedLateness > 0 окно может эмитить результат несколько раз — каждый раз с обновлёнными данными. Downstream должен правильно это обрабатывать.

events.keyBy(Event::getKey)
      .window(TumblingEventTimeWindows.of(Duration.ofMinutes(5)))
      .allowedLateness(Duration.ofMinutes(15))
      .reduce((a, b) -> a.merge(b));

Sequence:

1. t=10:05, wm=10:05 — окно [10:00, 10:05) закрывается. Trigger FIRE_AND_PURGE? Нет, с allowedLateness — FIRE без PURGE. Emit #1.
2. t=10:07, late event with ts=10:03 — добавлен в окно. Trigger FIRE. Emit #2 с обновлёнными данными.
3. t=10:12, late event with ts=10:04 — добавлен. Emit #3.
4. t=10:20, wm=10:20 — прошло window.end + allowedLateness (10:05 + 15min = 10:20). Окно окончательно очищается. Дальнейшие late events дропаются (или в side output).

Downstream должен:

• Upsert-aware sink. Обновление по window key + start вместо append. См. урок 05.05.
• Idempotent processing. Если downstream — это analytics, должен правильно обработать “перезапись” значения.
• Final emit detection. Опционально: помечать последний emit окна флагом is_final=true для downstream-логики типа alerting.

DLQ паттерн для late events

В production-jobs side output для late events часто маршрутизируется в Dead Letter Queue (DLQ) — отдельный Kafka топик для асинхронной обработки.

OutputTag<Event> lateTag = new OutputTag<Event>("late-events"){};

SingleOutputStreamOperator<Result> windowed = events
    .keyBy(Event::getKey)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(5)))
    .allowedLateness(Duration.ofMinutes(15))
    .sideOutputLateData(lateTag)
    .reduce(new MyReducer());

// Основной поток
windowed.sinkTo(kafkaSink);

// DLQ для late events
windowed.getSideOutput(lateTag)
    .map(e -> new LateEventRecord(e, System.currentTimeMillis()))
    .sinkTo(dlqKafkaSink);

DLQ-топик потом обрабатывается:

• Alerting. Алерт, если количество late events резко выросло (например, > 1000 за минуту).
• Backfill batch. Раз в день batch-job читает DLQ, агрегирует пропущенные данные, и записывает корректировки в analytics warehouse.
• Debug & analysis. Late events с большим опозданием могут указывать на проблемы upstream (медленный producer, network issues).

Дросселирование re-emissions

Re-emit окна при каждом late event может перегрузить downstream:

• Если в окне 1000 событий пришло late, и каждое триггерит FIRE, downstream получит 1000 эмиссий одного окна.
• Sink с upsert будет переписывать значение 1000 раз — нагрузка на network и database.

Решение — кастомный trigger с дросселированием:

public class ThrottledLateFireTrigger extends Trigger<Object, TimeWindow> {
    private final ValueStateDescriptor<Long> lastFiredDescriptor =
        new ValueStateDescriptor<>("last-late-fired", Long.class);
    private final long throttleMs = Duration.ofMinutes(1).toMillis();

    @Override
    public TriggerResult onElement(Object e, long ts, TimeWindow window, TriggerContext ctx) throws Exception {
        if (ts < window.maxTimestamp() && ctx.getCurrentWatermark() >= window.maxTimestamp()) {
            // late event
            ValueState<Long> lastFired = ctx.getPartitionedState(lastFiredDescriptor);
            Long last = lastFired.value();
            long now = ctx.getCurrentProcessingTime();
            if (last == null || now - last >= throttleMs) {
                lastFired.update(now);
                return TriggerResult.FIRE;
            }
            return TriggerResult.CONTINUE;
        }
        ctx.registerEventTimeTimer(window.maxTimestamp());
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) {
        if (time == window.maxTimestamp()) return TriggerResult.FIRE_AND_PURGE;
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) {
        return TriggerResult.CONTINUE;
    }

    @Override
    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
        ctx.getPartitionedState(lastFiredDescriptor).clear();
    }
}

Эта стратегия эмиттит late-update максимум раз в минуту: первый late event сразу триггерит FIRE, следующие 60 секунд только обновляют accumulator без emit.

Метрики и мониторинг

Стандартные Flink метрики для late events:

• numLateRecordsDropped per window operator — сколько событий выпало совсем (after allowed lateness).
• Watermark gap — разница между current processing time и current watermark.
• windowState-related метрики — размер state, количество активных окон.

Алерты в production:

• numLateRecordsDropped > 0 — любое количество late events может быть индикатором проблем upstream.
• Watermark gap > 2x boundedOutOfOrderness — watermark отстаёт сильнее ожидаемого. Часто индикатор idle partition.
• Резкий рост числа активных окон — может указывать на проблемы с emit или catastrophic late events.

Антипаттерны late event handling

Antipattern 1: огромный allowedLateness “на всякий случай”.

.allowedLateness(Duration.ofDays(7))  // ПЛОХО

Каждое окно живёт в state 7 дней. Для часовых окон это означает 168 одновременно активных окон в state per ключ. RAM/диск катастрофически растут. Правильно: разумный allowedLateness (минуты-часы) + DLQ для action на supersene-late.

Antipattern 2: отсутствие upsert downstream.

.allowedLateness(Duration.ofMinutes(15))
// ...
.sinkTo(plainKafkaSink);  // append-only, dust!

Late events дают re-emit, append-only sink накапливает дубликаты. См. урок 05.05.

Antipattern 3: дропать late events без observability.

Дефолтное поведение — silent drop. В production это означает, что вы не знаете о потере данных. Всегда sideOutputLateData + monitoring.

Antipattern 4: late events в processing-time окне.

Processing time windows вообще не знают о late events — они закрываются по wall-clock. События с старым event time будут просто отсчитываться в “текущее” окно неверно. Если ваш use-case требует обработки late events — переходите на event time.

Сценарии: outline для разных типов pipeline

Real-time fraud detection (latency-critical).

• boundedOutOfOrderness = 1 секунда.
• allowedLateness = 0.
• sideOutputLateData -> отдельный alert channel.
• Downstream: дополнительный pipeline для re-evaluation late events.

Hourly business metrics (correctness-critical).

• boundedOutOfOrderness = 30 секунд.
• allowedLateness = 1 час.
• Throttled trigger: re-emit не чаще раз в 10 минут.
• Sink: upsert into data warehouse.

Mobile app events (high lateness expected).

• boundedOutOfOrderness = 5 минут.
• allowedLateness = 2-4 часа.
• DLQ for super-late events (> 4 hour delay).
• Batch nightly reconciliation from DLQ.

Backfill / CDC initial snapshot.

• НЕ использовать allowedLateness в обычном смысле.
• См. урок 04 про withWatermarkAlignment.
• Альтернатива: bypass watermarks через ingestion time для backfill, переключиться на event time после catchup.

Production-чеклист

• Замерьте реальную задержку событий (p99) — на этом основывайте boundedOutOfOrderness.
• allowedLateness в порядке 5-20% от window length для большинства случаев.
• Всегда sideOutputLateData + DLQ-маршрут.
• Метрика numLateRecordsDropped мониторится, алертится при росте.
• Downstream sink с upsert-семантикой (Kafka compacted topic с правильным ключом или БД с ON CONFLICT UPDATE).
• Для long-tail late events рассмотрите throttling trigger вместо emit на каждое.

Попробуй сам

1. Multiple firings observation. В job-е с allowedLateness(30min) симулируйте late events с разными timestamps. Watch downstream — должны прийти multiple emissions для одного и того же окна.

2. DLQ side output. Установите allowedLateness(0) и sideOutputLateData. Отправьте late events. Прочитайте DLQ — должны быть все late events с их event timestamps.

3. Throttled emit. Реализуйте custom trigger из примера выше. Сравните количество emissions с обычным EventTimeTrigger при потоке 100 late events за минуту — throttled должен дать 1 emission вместо 100.