До сих пор мы рассматривали микросервисы как «конвейеры без памяти»: событие вошло, трансформировалось, вышло. Но реальный бизнес требует контекста. Чтобы ответить на вопрос «Какова общая сумма покупок этого клиента за последние 30 дней?» или «Сколько единиц товара осталось на складе после этой транзакции?», сервису нужно помнить. Ему нужно состояние.

Эта глава погружается в самую сложную и одновременно самую мощную часть событийно-управляемой архитектуры — потоковую обработку с поддержкой состояния (Stateful Stream Processing). Мы разберём, как состояние рождается из потока событий, как оно защищается от потери при сбоях, какие типы хранилищ существуют и почему дубликаты — это не баг, а неизбежная реальность распределённых систем.

Это седьмая глава цикла. Она опирается на главу II (table-stream duality, CQRS), главу V (state stores, changelog, standby replicas) и главу VI (event time, out-of-order). Здесь мы углубляем каждую из этих тем.

IДуальность «Поток — Таблица»: философский камень EDM

В основе stateful-обработки лежит одна из самых элегантных концепций событийной архитектуры: поток событий и таблица состояний — это два представления одной и той же сущности.

Эта дуальность работает в обе стороны. Любую таблицу можно превратить в поток, записывая каждое её изменение как событие (паттерн Change Data Capture). И любой поток можно превратить в таблицу, последовательно применяя все его события (процесс материализации).

IIChangelog и checkpoint: страховка от амнезии

Когда микросервис материализует состояние в локальное хранилище (RocksDB на диске), он обретает скорость: чтение и запись происходят локально, без сетевых задержек. Но если сервер сгорит, локальное состояние будет утеряно.

Решение — журнал изменений (changelog topic). Это специальный compacted Kafka-топик, куда микросервис записывает каждое изменение своего внутреннего состояния. По сути, это «страховочная копия» таблицы, разложенная на последовательность фактов.

Input Topic → Processor → State Store (RocksDB, local disk)
                    ↓
              Changelog Topic (Kafka, compacted)

Механизм восстановления. При потере инстанса новый инстанс подписывается на changelog и «проигрывает» все события с начала, восстанавливая RocksDB. Поскольку changelog — compacted topic, хранится только последнее значение для каждого ключа, поэтому объём восстановления пропорционален размеру состояния, а не количеству обработанных событий.

Checkpoint file

Kafka Streams дополнительно ведёт локальный checkpoint-файл, отслеживающий последний offset, для которого состояние сохранено. При рестарте приложение сначала проверяет checkpoint: если он есть, восстановление начинается с последней позиции, а не с нуля.

IIIRocksDB под капотом: LSM-дерево и практические последствия

RocksDB — embedded key-value движок, построенный на LSM-дереве (Log-Structured Merge Tree). Понимание его устройства необходимо для диагностики проблем производительности и роста состояния.

Write → MemTable (in-memory, sorted)
           ↓ flush
         SSTable (immutable file on disk)
           ↓ compaction
         Larger SSTables (merged, deduplicated)

1. Запись попадает в in-memory MemTable — отсортированный буфер.
2. При заполнении MemTable flush'ится на диск как SSTable — неизменяемый отсортированный файл.
3. Фоновая compaction сливает несколько SSTable в более крупные, удаляя дубликаты и освобождая место.

Tombstone: удаление, которого нет

При удалении ключа RocksDB не удаляет запись физически — записывается специальный маркер tombstone. Во время чтения RocksDB видит tombstone и не возвращает значение. Но сама запись и tombstone продолжают занимать место в SSTable-файлах до compaction.

// Принудительный compaction каждый час (дефолт = 0, выключен!)
options.setPeriodicCompactionSeconds(3600);
// Удаление obsolete файлов каждый час
options.setDeleteObsoleteFilesPeriodMicros(3600 * 1000000L);

Bloom filters

RocksDB по умолчанию использует Bloom filters для ускорения point lookups. Это критически важно для дедупликации: проверка «был ли уже этот event_id» должна быть быстрой. Без Bloom filters каждая проверка сканировала бы несколько SSTable.

IVШесть типов state stores: не один размер для всех

Kafka Streams предоставляет шесть типов хранилищ состояний, каждый заточенный под конкретную задачу.

Persistent KeyValueStore

Дефолт. Хранит последнее значение на ключ. Растёт бесконечно: нет встроенного TTL. Используется для счётчиков, дедупликации, материализации CQRS read models.

Window Store

Для оконных агрегаций. Сегменты старше retention автоматически удаляются, changelog использует compact,delete политику. Ограничение размера — встроенное.

Session Store

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

Global Store

Полная репликация всех партиций на каждый инстанс. Не требует соподразделения. Подходит для компактных справочников.

VVersioned store и KIP-889: время в state store

Versioned KeyValueStore (KIP-889, Kafka 3.5+) — революция для stream-table joins. Вместо одного значения на ключ хранит несколько версий с timestamp.

Price of product X:
  10:00 → $10
  10:05 → $15
  10:03 → $12   ← out-of-order!

Versioned store:
  get("X", asOfTimestamp=10:03) → $12  ← correct!
  get("X", asOfTimestamp=10:05) → $15
  get("X", asOfTimestamp=10:08) → $15  ← latest

Это решает проблему out-of-order в stream-table joins: если событие 10:03 пришло позже события 10:05, versioned store корректно вернёт состояние на 10:03.

// Создание versioned store (Kafka 3.5+)
StoreBuilder<VersionedKeyValueStore<String, Price>> priceStore =
    Stores.versionedKeyValueStoreBuilder(
        Stores.persistentVersionedKeyValueStore(
            "prices-store",
            Duration.ofDays(7)  // history retention = 7 days
        ),
        Serdes.String(), Serdes.forPrice()
    );

History retention одновременно является grace period: события с timestamp старше текущего stream time минус retention — отбрасываются.

VIInteractive Queries: IQv1 и IQv2

State store — не только внутренний кэш для обработчика. Его можно читать снаружи через REST API — это заменяет отдельную базу данных для query-сценариев.

IQv1 — запрос локального state store

ReadOnlyKeyValueStore<String, Long> store = streams.store(
    StoreQueryParameters.fromNameAndType(
        "word-counts",
        QueryableStoreTypes.keyValueStore()
    )
);
Long count = store.get("kafka");  // point lookup

Состояние распределено по партициям, каждый инстанс хранит только свою долю. Для запроса «по всему состоянию» нужен RPC-слой: инстанс, получивший запрос, определяет, на каком инстансе живёт нужный ключ (через KafkaStreams#metadataForKey), и либо отвечает локально, либо проксирует запрос на удалённый инстанс.

IQv2 — composable Query-объекты (KIP-796)

StateQueryRequest<Long> request = StateQueryRequest.inStore("word-counts")
    .withQuery(KeyQuery.withKey("kafka"));
StateQueryResult<Long> result = streams.query(request);

IQv2 возвращает StateQueryResult с per-partition результатами и информацией о позиции (Position) — позволяет оценить свежесть данных. Поддерживает KeyQuery (point lookup) и RangeQuery (сканирование).

Stale stores

enableStaleStores() (KIP-535) — чтение из standby-реплик во время восстановления. Жертвует свежестью ради доступности: «лучше ответ, отстающий на секунду, чем исключение при каждом деплое».

VIITTL-проблема и KIP-1225

KeyValueStore растёт бесконечно. Нет встроенного TTL, нет eviction thread, нет expireAfter. Каждое значение, записанное в plain KV store, остаётся навсегда — через рестарты, через rebalance — пока вы не удалите его вручную.

Почему window stores истекают, а KV — нет

WindowStore автоматически удаляет сегменты старше retention, а его changelog использует compact,delete политику. KeyValueStore использует compact-only changelog: tombstone никогда не записывается автоматически, поэтому при восстановлении все «удалённые» ключи воскресают из changelog.

Три рабочих паттерна очистки

1. Re-key в WindowStore. Переподразделить данные по entity-ключу в WindowStore с retention = желаемый TTL. Сегменты истекают автоматически, changelog — тоже. Нет punctuator, нет ручных tombstones, нет риска воскрешения.
2. Punctuator + ручной tombstone. Processor API: периодически сканировать store, удалять устаревшие ключи и записывать tombstone в changelog (не просто store.delete(), а context.forward(key, null)). Без tombstone восстановление из changelog вернёт «удалённые» данные.
3. TimestampStore + пакетное удаление. Отдельный store, хранящий timestamp последнего изменения каждого ключа. Периодически запускать очистку: найти ключи старше TTL, пакетно удалить из всех связанных stores, записать tombstones.

Пример punctuator для очистки

public void init(ProcessorContext context) {
    this.context = context;
    this.dedupStore = (KeyValueStore<String, Long>) context.getStateStore("dedup");
    context.schedule(
        Duration.ofMinutes(30),
        PunctuationType.WALL_CLOCK_TIME,
        this::cleanupOldEntries
    );
}

void cleanupOldEntries(long timestamp) {
    long cutoff = timestamp - Duration.ofHours(24).toMillis();
    try (KeyValueIterator<String, Long> it = dedupStore.all()) {
        while (it.hasNext()) {
            var entry = it.next();
            if (entry.value < cutoff) {
                dedupStore.delete(entry.key);
                context.forward(entry.key, null);  // tombstone в changelog
            }
        }
    }
}

KIP-1225: будущее TTL

KIP-1225 (август 2025) — предложение добавить нативный TTL в TimestampedKeyValueStore через StoreBuilder.withTtl(Duration). Автоматическое истекание по embedded timestamp, tombstone-flush в changelog, eviction по wall-clock time. Пока не реализовано, но закрывает самую частую боль.

VIIIГорячие реплики: мгновенное восстановление

Восстановление из changelog — надёжное, но не мгновенное. Если состояние занимает гигабайты, replay может занять минуты. Для критичных процессов это неприемлемо.

Горячие реплики (Standby Replicas). Пока инстанс А активно обрабатывает партицию P0, инстанс Б в фоновом режиме тоже читает тот же changelog P0 и строит идентичную копию состояния. При отказе А контроллер мгновенно переключает обработку на Б — состояние уже готово.

Настройка: num.standby.replicas=1 — одна резервная реплика на каждую активную задачу. Время восстановления сокращается с минут до секунд. Цена — двойной расход диска.

IXStream-Table Joins: обогащение в реальном времени

Одна из самых мощных операций — обогащение потока событий данными из таблицы состояний в реальном времени.

Для корректной работы оба источника должны быть соподразделены: одинаковое число партиций и один partitioner. JOIN происходит локально, в памяти — без сетевых вызовов.

Таблица в stream-table join — живая. Она постоянно обновляется из своего потока событий. Если профиль изменился, KTable это отразит, и последующие JOIN-ы будут использовать свежие данные.

Проблема классического KV store

Обычный KeyValueStore хранит только последнее значение на ключ. Если событие 10:03 пришло после события 10:05 (out-of-order), JOIN использует уже обновлённое значение, а не то, что было на 10:03 — результат некорректен. Versioned store (KIP-889) решает: get(key, 10:03) возвращает значение, актуальное на 10:03.

XEffectively-once и дедупликация

Гарантировать, что каждое событие будет обработано строго один раз (exactly-once), математически невозможно в общем случае. Сети рвутся, таймауты срабатывают, и продюсер не знает, было ли событие записано. Поэтому индустрия оперирует термином «практически однократная обработка» (effectively-once).

Транзакции «Клиент — Брокер» (exactly_once_v2)

Kafka Streams с версии 2.6 (rename на exactly_once_v2 в 3.0) поддерживает атомарные транзакции: чтение входного события, обновление state store, запись в output topic и коммит offset — в одной Kafka-транзакции. Либо всё фиксируется, либо всё откатывается.

processing.guarantee=exactly_once_v2

v1 (exactly_once) был deprecated в Kafka 3.0 и удалён в 4.0 — требовал одного продюсера на каждую входную партицию, что создавало огромный overhead. v2 — единственный рекомендуемый режим.

Идемпотентность на стороне потребителя

Даже с транзакциями дубликаты могут проникнуть (если продюсер повторил отправку до получения ack). Потребитель должен распознавать и игнорировать дубликаты:

1. Уникальный ID — каждое событие получает event_id (UUID или бизнес-ключ + версия).
2. Таблица дедупликации — materialized store обработанных ID. Перед обработкой — проверка.
3. Атомарная проверка-и-запись — проверка дубликата и обновление бизнес-состояния в одной операции.

public void process(Record<String, OrderEvent> record) {
    String eventId = record.value().getEventId();
    if (dedupStore.get(eventId) != null) {
        return;  // уже обработано
    }
    // Бизнес-логика
    long newTotal = currentTotal + record.value().getAmount();
    balanceStore.put(record.key(), newTotal);
    // Запись в дедуп-таблицу — в том же state store, атомарно с бизнес-обновлением
    dedupStore.put(eventId, System.currentTimeMillis());
}

Поддержание согласованного состояния при сбоях

Самый коварный сценарий: микросервис обработал событие, обновил state store и таблицу дедупликации, но не успел закоммитить offset. При перезапуске он прочитает то же событие заново.

Если бизнес-операция неидемпотентна (balance += 100), повторная обработка удвоит сумму. Именно поэтому:

• Обновление состояния и фиксация offset должны быть атомарны (exactly_once_v2).
• Таблица дедупликации хранится в том же state store, что и бизнес-данные, и восстанавливается из того же changelog.

XIИмя state store — это API decision

Имя state store не просто идентификатор. Оно попадает в:

• Имена internal topics (changelog, repartition)
• Метрики и дашборды
• Interactive query код
• ACLs и governance tooling
• Runbooks и alerting

Переименование store = смена changelog topic = полная потеря состояния при следующем рестарте.

// Слабое имя — implementation detail
Materialized.as("store1")

// Сильное имя — domain projection
Materialized.as("account-payment-total")

XIIМиграция и name collision

State store name collision

Если два processor'а используют store с одинаковым именем — Kafka Streams выбросит TopologyException. Это частая ошибка при рефакторинге: переименовали store в одном месте, забыли в другом.

Правило: имена store должны быть уникальны в пределах всей топологии. Для избежания конфликтов используйте префиксы по домену:

Materialized.as("account.balance")
Materialized.as("account.dedup")
Materialized.as("inventory.stock")
Materialized.as("inventory.dedup")

State store migration: как обновить топологию

Kafka Streams не поддерживает изменение топологии без сброса store. Если вы изменили логику обработки — старое состояние может стать несовместимым.

XIIIМониторинг stateful приложения

Stateful приложение — это не «настроил и забыл». Без мониторинга вы узнаете о проблеме только когда диск переполнится или восстановление затянется на часы.

Метрики RocksDB

Changelog lag

Ключевая метрика: record-lag-max по changelog topic. Если lag растёт — восстановление займёт больше времени. Алерт при lag > 100k записей или при росте за 15 минут.

Время восстановления

Замеряется через JMX или custom metric. Практический способ: имитировать падение инстанса в staging и замерить, сколько времени занимает полное восстановление. Это значение должно укладываться в SLA.

Compaction lag

Если compaction отстаёт — SSTable-файлы копятся, диск переполняется. Метрика: rocksdb-compaction-pending-bytes. Алерт при > 10 GB или при росте за час.

XIVFlink vs Kafka Streams: две модели состояния

Flink 2.0: disaggregated state backend (ForSt)

Flink 2.0 (март 2025) представил disaggregated state backend (ForSt): состояние живёт в remote object storage (S3), а не на локальном диске TaskManager. State pages in/out по мере необходимости.

Что это даёт:

• Decouples state size от диска. Не нужно выделять SSD под каждый TaskManager.
• Ускоряет rescaling. TaskManagers просто начинают читать другие key partitions из S3.
• Упрощает эксплуатацию. State — это просто объекты в S3, а не тома на локальных дисках.

Trade-offs:

• Latency. Page in/out из S3 занимает десятки миллисекунд, а не микросекунды, как локальный RocksDB.
• Стоимость. Оплата за S3 API calls + storage.
• Не для всех workloads. Для latency-sensitive задач (real-time fraud detection) лучше embedded RocksDB.

Когда выбирать:

• Kafka Streams — pipeline Kafka-to-Kafka, moderate state (до 50–100 GB), нужна простота деплоя.
• Flink — большой state (терабайты), сложная аналитика, multi-source, savepoints для zero-downtime upgrades, ML-пайплайны.
• Flink + disaggregated state — state > 1 TB, elastic rescaling, batch + streaming unified.

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

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

Практический чеклист для stateful приложения

• State store назван осмысленно (account-payment-total, а не store1)
• Выбран правильный тип store (KV для latest, window для агрегаций, versioned для temporal joins)
• Для бесконечно растущего KV store реализована очистка (punctuator + tombstone, или re-key в window store)
• RocksDB настроен: periodicCompactionSeconds, deleteObsoleteFilesPeriodMicros
• Changelog topic имеет то же число партиций, что и state store
• num.standby.replicas >= 1 для критичных stateful сервисов
• processing.guarantee=exactly_once_v2 для business-critical пайплайнов
• Таблица дедупликации живёт в state store, а не во внешней БД
• Interactive Queries настроены с RPC-слоем для distributed state
• Имена store уникальны в пределах топологии (префиксы по домену)
• При breaking changes — новый store name вместо переименования
• Мониторинг: state store size, RocksDB compaction lag, changelog lag
• Проверено: время восстановления после потери local state приемлемо

Чеклист мониторинга

• rocksdb-block-cache-hit-ratio > 0.8
• rocksdb-estimated-num-keys — не растёт линейно без TTL
• rocksdb-compaction-pending-bytes < 10 GB
• record-lag-max changelog < 100k записей
• Время восстановления измерено и укладывается в SLA
• Алерты на рост state store size настроены

Глава VII цикла «Событийно-управляемая архитектура». Источники: Apache Kafka, Confluent, Conduktor, Automq, Lydtech Consulting, Habr (кейс «Магнита»), RisingWave, Software Patterns Lexicon, KIP-889, KIP-796, KIP-535, KIP-1225.