В идеальном мире события прибывают в брокер и покидают его строго в том порядке, в котором они были созданы. Часы на всех серверах синхронизированы до наносекунды, сеть никогда не рвётся, а потребители обрабатывают данные мгновенно. Но мы живём в реальности. Сетевые пакеты застревают, серверы перезагружаются, а производители событий временно теряют связь с брокером.

Как гарантировать, что микросервис, обрабатывающий финансовые транзакции или аналитику кликов, выдаст один и тот же результат независимо от того, обрабатывает ли он данные в реальном времени или «переваривает» вчерашний завал? Ответ кроется в концепции детерминированной обработки потоков.

Это шестая глава цикла о событийно-управляемой архитектуре. Мы уже разобрали, почему EDA нужна, как она устроена под капотом, как проектировать контракты, как подключить legacy и как работает потоковая обработка. Теперь — о самом сложном: о времени.

IИллюзия времени: событие vs обработка

Главная ловушка при переходе от пакетной обработки к потоковой — путаница между двумя типами времени. Чтобы построить надёжную систему, нужно жёстко разграничить:

IITimestamp Assigner: откуда система берёт event time

Прежде чем двигаться дальше, нужно ответить на фундаментальный вопрос: откуда система знает event time? Событие — это просто байты. Время нужно откуда-то извлечь.

Kafka Streams

По умолчанию Kafka Streams использует timestamp записи Kafka — то, что продюсер положил в metadata. Это может быть время отправки (create time) или время приёма брокером (log append time) — зависит от настройки топика message.timestamp.type.

Если timestamp Kafka не подходит (например, событие реплицировано из другой системы, и метка не отражает реальное время), можно задать свой TimestampExtractor:

public class OrderTimestampExtractor implements TimestampExtractor {
    @Override
    public long extract(ConsumerRecord<Object, Object> record, long partitionTime) {
        OrderEvent order = (OrderEvent) record.value();
        return order.getOccurredAt().toEpochMilli();
    }
}

// В конфигурации:
props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
          OrderTimestampExtractor.class);

Особенно важно: если payload не содержит корректного timestamp, продюсер может положить отрицательное значение. Тогда Kafka Streams выбросит StreamsException. Для graceful-обработки используйте WallclockTimestampExtractor (fallback на текущее время) или FailOnInvalidTimestamp (для строгой валидации).

Apache Flink

В Flink event time извлекается через TimestampAssigner, который задаётся на источнике:

DataStream<OrderEvent> orders = env
    .addSource(new KafkaSource<>(...))
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(
                Duration.ofMinutes(2))
            .withTimestampAssigner((event, ts) ->
                event.getOccurredAt().toEpochMilli()));

Ключевое отличие от Kafka Streams: во Flink timestamp assigner и watermark strategy задаются явно. Это требует больше кода, но даёт полный контроль над моделью времени.

IIIПланировщик событий: дирижёр оркестра

Микросервис часто потребляет данные из нескольких партиций и разных потоков одновременно. Если обрабатывать по принципу «что первое пришло, то и крутим», результат зависит от сетевой задержки, а не от бизнес-логики.

Планировщик событий (Event Scheduler) буферизует входящие записи и всегда отдаёт в бизнес-логику событие с наименьшей временной меткой. Это гарантирует, что даже если события перепутались на уровне сети, микросервис восстановит их историческую справедливость.

Цена: буферизация требует памяти. Чем больше окно и чем больше партиций, тем больше буфер. Для session windows с timeout 30 минут и 100 партиций буфер может занимать гигабайты.

IVВодяные знаки: как понять, что «прошлое» уже прошло

Если мы буферизуем события для сортировки, возникает вечный вопрос: как долго ждать опоздавших? Бесконечно ждать нельзя — память не резиновая, а бизнесу нужны результаты.

Водяной знак (Watermark) — это монотонно возрастающий timestamp, который объявляет: «Все события с временем T и ранее уже гарантированно прибыли». Если после водяного знака приходит событие со временем T−1, система понимает: это запоздалое событие (Late Event).

Водяной знак вычисляется на основе наблюдаемых временных меток событий. Стратегия генерации определяет, насколько агрессивно система «сдвигает» границу.

Periodic Watermark

Водяной знак генерируется периодически (например, каждые 200 мс), на основе максимального event time, наблюдённого за период, минус допустимая задержка:

watermark = max(eventTime) - allowedLateness

Например, если максимальное наблюдённое event time — 10:05, а allowed lateness — 2 минуты, watermark = 10:03. Все окна до 10:03 можно закрывать.

Punctuated Watermark

Водяной знак генерируется при каждом специальном событии (например, маркер конца сессии). Редко используется, подходит для потоков с нерегулярными интервалами.

Пример настройки во Flink

WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(
        Duration.ofSeconds(30))
    .withTimestampAssigner((event, ts) -> event.getOccurredAt().toEpochMilli())
    .withIdleness(Duration.ofMinutes(5));

Здесь forBoundedOutOfOrderness означает: «мы ожидаем, что события опаздывают не более чем на 30 секунд».

VWatermark propagation и watermark alignment

Watermark движется по топологии: каждый оператор получает водяные знаки от всех входных потоков и берёт минимум.

Source A: watermark = 10:05
Source B: watermark = 10:02  →  Operator: watermark = min(10:05, 10:02) = 10:02
Source C: watermark = 10:07

Это объясняет, почему один медленный или молчащий поток блокирует всю топологию: если один источник не шлёт события, его watermark не двигается, и общий минимум застывает.

Watermark alignment — механизм «притормаживания»

В больших топологиях разные источники могут обрабатывать данные с разной скоростью. Если один источник отстаёт на часы, а остальные — на секунды, минимум watermark застывает, и окна не закрываются.

Flink предоставляет watermark alignment: операторы могут «сообщать» upstream, что они отстают, и upstream-операторы притормаживают свои watermarks, чтобы дать отстающему догнать.

WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(
        Duration.ofSeconds(30))
    .withWatermarkAlignment(
        "alignment-group-1",
        Duration.ofMinutes(5),   // максимальное расхождение
        Duration.ofSeconds(30)); // интервал проверки

Это более продвинутый механизм, чем withIdleness: он не исключает источник, а временно «придерживает» остальные, чтобы сохранить консистентность.

VIПроблема простаивающих источников (Idle Sources)

Если одна партиция замолчала (например, ночной трафик по одному региону прекратился), её watermark перестаёт двигаться. Поскольку оператор берёт минимум по всем входам, общая watermark застывает, и окна не закрываются.

Решение в Flink: withIdleness(Duration.ofMinutes(5)) — если источник не получает событий 5 минут, он исключается из расчёта watermark. Окна продолжают закрываться по остальным источникам.

В Kafka Streams этой проблемы нет в явном виде — окна закрываются по wall-clock time, а не по watermark. Но есть другая проблема: если партиция замолчала, окно не получит новых событий и будет закрыто по истечении grace period.

VIIОкна: нарезка времени на куски

Бизнес-логика редко работает с абстрактным «потоком». Ей нужны агрегаты: «сколько кликов за последний час», «какова сессия пользователя». Для этого используются оконные функции.

Переворачивающиеся (Tumbling) — фиксированные, непересекающиеся интервалы. Каждое событие попадает ровно в одно окно. Идеально для периодических отчётов: «выручка за каждый час».

Прыгающие (Hopping) — фиксированный размер, но шаг меньше размера. Окно «перекатывается» с заданным интервалом. Одно событие может попасть в несколько окон. Используется для rolling-метрик: «5-минутное скользящее среднее, обновление каждую минуту».

Скользящие (Sliding) — окно привязано не к фиксированной сетке, а к временным меткам самих событий. Два события агрегируются, если их временные метки в пределах размера окна. В Kafka Streams sliding windows используются только для JOIN-операций. Идеально для корреляции: «дважды неудачный логин в течение 30 секунд».

Пример настройки в Kafka Streams

TimeWindows windows = TimeWindows
    .ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1));

stream.groupByKey()
    .windowedBy(windows)
    .count();

Здесь окно — 5 минут, grace period — 1 минута: опоздавшие события в пределах 1 минуты после закрытия окна будут приняты и вызовут корректирующий результат.

VIIISession windows и механизм слияния

Сеансовые окна (Session Windows) — динамические окна, не привязанные к часам. Окно открыто, пока пользователь активен. Пауза (timeout бездействия, например 30 минут) закрывает окно.

Ключевая особенность session windows — слияние (merge). Если новая активность «дотягивается» до предыдущей (gap < timeout), два окна объединяются в одно.

Session 1: [10:00 — 10:05]
Session 2: [10:07 — 10:15]
Gap = 2 минуты < timeout (30 минут)
→ Объединённая сессия: [10:00 — 10:15]

Это основа для анализа пользовательского поведения: клики, просмотры, покупки в рамках одной сессии могут быть «склеены» в единое событие.

В Kafka Streams session windows используются так:

stream.groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(
        Duration.ofMinutes(30)))
    .count();

IXOut-of-order vs Late: две разные проблемы

Не все «неправильные» события одинаковы.

Различие критично: out-of-order — штатная ситуация, late — исключительная. Система должна быть спроектирована так, чтобы out-of-order обрабатывались прозрачно, а late — осознанно.

XТри стратегии борьбы с запоздалыми событиями

Рано или поздно watermark уйдёт вперёд, и прилетит событие, которое «опоздало». Что с ним делать — это не только технический, но и бизнес-вопрос.

Отбросить (Drop)

Окно закрыто, результат ушёл. Позднее событие игнорируется. Подходит для телеметрии, где важна актуальность, а не история.

В Kafka Streams — режим по умолчанию при отсутствии grace period. В Flink — события уходят в side output для анализа.

Льготный период (Grace Period / Allowed Lateness)

Окно закрыто, но остаётся «приоткрытым» на N минут. Если опоздавшее событие приходит в этот период, агрегат пересчитывается и выпускается корректирующий результат.

Разница между фреймворками:

• Kafka Streams: grace period — единый буфер до финального закрытия.
• Flink: allowed lateness — окно может переоткрываться многократно, пока allowed lateness не исчерпан.

Перенаправить (Side Output)

Позднее событие не отбрасывается молча, а направляется в отдельный поток (side output). Этот поток можно анализировать, репроцессить или применять fallback-логику.

OutputTag<OrderEvent> lateOutputTag = new OutputTag<>("late-orders"){};

SingleOutputStreamOperator<Result> result = stream
    .keyBy(OrderEvent::getCustomerId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .allowedLateness(Time.minutes(1))
    .sideOutputLateData(lateOutputTag)
    .aggregate(new OrderAggregator());

DataStream<OrderEvent> lateOrders = result.getSideOutput(lateOutputTag);

XIВыбор размера grace period: latency vs completeness

Grace period — это компромисс между задержкой выдачи результата (latency) и полнотой данных (completeness). Как выбрать?

1. Анализируйте distribution of delay. Если 95% событий приходят в течение 30 секунд, grace period 1 минуты покроет 95% случаев. Grace period 5 минут покроет 99.9%, но добавит 5 минут latency.
2. Различайте критичность. Для real-time дашбордов (клики, трафик) — grace 30 секунд, completeness 95%. Для биллинга — grace 10 минут, completeness 99.99%. Для ежедневной финансовой отчётности — grace 24 часа, completeness 100%.
3. Учитывайте cost of reprocessing. Если позднее событие отброшено, но данные важны — всегда можно reprocessing-ом. Grace period = буфер между «достаточно быстро» и «достаточно точно».
4. Измеряйте. Метрика для мониторинга: late-events-rate — доля событий, пришедших после watermark. Если растёт — grace period нужно увеличить.

XIIFlink vs Kafka Streams: две модели времени

Фреймворки принципиально расходятся в подходе к времени, и это влияет на выбор под задачу.

Flink — когда нужна точная event-time обработка с watermarks, сложная аналитика, multi-source JOIN, ML-пайплайны. Kafka Streams — когда вход и выход в Kafka, нужна простота деплоя и приемлем wall-clock fallback.

XIIIСуперсила EDM: reprocessing

Неизменяемые журналы событий дают потрясающую возможность. Если найдена ошибка в бизнес-логике или выпущена новая схема, можно просто перемотать offset на начало и прогнать данные через микросервис заново. Благодаря детерминированной обработке и планировщику событий, переработка исторических данных выдаст точно такой же порядок и результаты, как если бы обработка шла в реальном времени.

Но reprocessing — не «нажать кнопку». Нужно учесть:

1. Очистить state store. Перед reprocessing локальный RocksDB нужно сбросить, иначе агрегаты будут некорректно наложены на старое состояние.
2. Output topic. Либо очистить выходной топик, либо писать в новый. Запись в тот же топик поверх старых данных создаст дубликаты.
3. Schema evolution. События годичной давности могут быть записаны по старой схеме. Новая версия кода может не десериализовать их. Решение — upcasting: функция-преобразователь, которая читает старую версию события и переводит в новую in-memory.
4. Побочные эффекты. Если топология включает отправку email, вызов стороннего API или платёжного шлюза, reprocessing приведёт к катастрофе: клиенты получат дублирующие письма за прошлые годы. Перед reprocessing всегда отключайте или ставьте на заглушки все неидемпотентные внешние вызовы.

Пример upcasting

public OrderEvent upcast(OrderEventV1 old) {
    return new OrderEvent(
        old.getId(),
        old.getCustomerId(),
        old.getAmount(),
        null,                       // paymentMethod — новое поле
        old.getOccurredAt()
    );
}

Оценка времени reprocessing

Reprocessing 1 ТБ данных через Kafka Streams при 50k msg/sec ≈ 30–60 минут. Но это очень приблизительно. Ключевые факторы:

• Bottleneck может быть в state store, а не в обработке. RocksDB на медленном диске читает в 10 раз медленнее, чем на NVMe.
• Batch size — большие батчи ускоряют throughput, но увеличивают memory pressure.
• Параллелизм — ограничен числом партиций. Если партиций 6, а вы запустили 12 инстансов — 6 простаивают.

Чеклист перед reprocessing

• State store очищен
• Output topic — новый или очищен
• Все внешние вызовы (email, SMS, платёжные API) отключены или идемпотентны
• Schema upcasters настроены для старых версий событий
• Offset сброшен на начало (или на нужную позицию)
• Мониторинг lag включён — reprocessing создаёт всплеск нагрузки
• Оценено время: reprocessing 1 ТБ данных ≈ 30–60 минут при 50k msg/sec
• Диск под state store имеет достаточный запас

XIVЧеклисты и связь с циклом

Связь с предыдущими главами

Чеклист по настройке grace period

• Distribution of delay измерен (95-й, 99-й перцентиль)
• Критичность данных определена (real-time vs reporting)
• SLA на latency зафиксирован
• Cost of late data оценён (audit, fallback)
• Fallback-стратегия определена (side output, drop, replay)
• Метрика late-events-rate добавлена в мониторинг
• Idle sources настроены (withIdleness во Flink)
• Watermark alignment настроен, если источники разной скорости

Ключевые выводы главы

1. Event time — единственное детерминированное время. Processing time зависит от инфраструктуры и непредсказуем.
2. Timestamp assigner определяет, откуда система берёт event time — из payload или из metadata Kafka.
3. Планировщик событий переупорядочивает out-of-order события по event time. Это штатная работа, не ошибка.
4. Водяные знаки (Flink) и grace period (Kafka Streams) — механизмы определения границы «прошлое прошло».
5. Watermark propagation — минимум по всем входам. Watermark alignment — механизм «притормаживания» для отстающих источников.
6. Idle sources могут заморозить watermark. Flink решает через withIdleness, Kafka Streams — через wall-clock fallback.
7. Late events — три стратегии: drop, grace period (корректирующий результат), side output (audit/fallback).
8. Session windows сливаются при gap < timeout. Это основа для анализа пользовательского поведения.
9. Flink vs Kafka Streams — две модели времени. Flink: event-time-first, watermarks, alignment. Kafka Streams: wall-clock + grace.
10. Reprocessing — суперсила неизменяемых журналов, но требует очистки state store, нового output topic, upcasting схем и отключения побочных эффектов.

Глава VI цикла «Событийно-управляемая архитектура». Источники: Apache Flink, Confluent, Conduktor, RisingWave, Software Patterns Lexicon, BigDataSchool, Habr, практические кейсы production-систем.