Сразу: проект сделан с помощью ИИ. Если вы против, статью можно пропустить.
Сам не люблю иишный слоп. Берусь только за то, что мне реально мешало/мешает, и все проверяю сам. Хотел попробовать kafka streams в проде с node.js + typescript, но не нашел хорошего решения, поэтому сделал сам.
Кому это нужно
Если у вас стек целиком на TypeScript, Kafka или вы уже читаете через @platformatic/kafka или KafkaJS, и захотелось Kafka Streams — это для вас. Состояние по ключу (счетчики, агрегаты, окна) обычно живет в Redis или Postgres, и при падении консьюмера посреди батча получаете дубли или потерянные обновления. В Java для этого есть Kafka Streams. В Node.js аналога нет, а поднимать отдельный JVM‑сервис ради этого дорого.
kstreams‑ts переносит модель Kafka Streams в Node.js, поверх @platformatic/kafka. Состояние живет рядом с обработкой и переживает падения и ребалансы, а в режиме exactly‑once каждая запись учитывается ровно один раз.
Как это выглядит
Считаем просмотры страниц по пользователю и пишем каждое новое значение счетчика в другой топик:
import { createStreamsApp, serdes } from 'kstreams-ts'const app = createStreamsApp({ applicationId: 'page-views', brokers: ['localhost:9092'], stateDir: './state', guarantee: 'exactly-once', changelogReplicationFactor: 1, // один локальный брокер; по умолчанию 3})app .stream('page-views', { key: serdes.string, value: serdes.json<{ path: string }>() }) .filter((_userId, value): value is { path: string } => value !== null) .groupByKey() .count({ store: 'views-per-user' }) .toStream() .to('views-per-user', { key: serdes.string, value: serdes.number })await app.start()
Типы проходят через всю цепочку: если тип ключа или значения на выходе не тот, код не скомпилируется. Кроме DSL (filter, map, groupByKey, count, aggregate, reduce, tumbling‑окна) есть Processor API для своей логики. Юнит‑тесты гоняют топологию в памяти, брокер не нужен.
Как устроено
Модель та же, что в Kafka Streams. Task — один на каждый номер партиции входных топиков, он владеет своими сторами. Worker — участник consumer group, Task’и переезжают вместе с его партициями. Worker’ы запускаются как worker threads или как отдельные процессы с тем же applicationId.
State Store — локальная SQLite. Каждая запись дублируется в compacted Changelog‑топик, и после падения или переезда Task’а стор восстанавливается из него. Checkpoint рядом со стором помнит offset, до которого стор совпадает с Changelog, так что на старте его не надо перечитывать целиком.
Клиент Kafka — @platformatic/kafka. Совместимости с Java Kafka Streams по API и форматам нет, я переносил идеи, а не код.
Exactly‑once
Выходные записи, записи в Changelog и offset’ы прочитанного уходят в одной Kafka‑транзакции. Записи в SQLite фиксируются только после ее коммита, а при отмене откатываются.
Самое сложное было с fencing: зомби не должен ничего закоммитить. Зомби тут — Worker, который завис на GC‑паузе, пропустил ребаланс и не знает, что его Task уже отдали другому. Защита на двух уровнях.
Первый — в брокере, KIP-447. Offset’ы коммитятся вместе с generation consumer group, и coordinator отклоняет коммит от участника, которого в группе уже нет. Если Task только писал по таймеру и ничего не читал, Worker заново коммитит его последний offset, чтобы проверка все равно сработала.
Второй — в самом Worker’е. У каждого Task’а есть эпоха владения, она меняется при каждой потере партиции. Перед коммитом Worker сверяет эпохи всех Task’ов в транзакции. Если изменилась хоть одна, транзакция отменяется.
Новый владелец ждет, пока закончится транзакция предыдущего, и только потом восстанавливает стор. А чтобы медленный Processor не уперся в transaction.timeout.ms, Worker подстраивает размер батча под темп обработки.
Как я это проверял
Юнит‑тестами такое не доказать, поэтому есть crash suite. Он запускает stateful‑приложение и убивает его SIGKILL в случайные моменты, пока на вход идут данные. Потом перезапускает и сверяет выходы, SQLite и Changelog’и с эталоном, посчитанным по входу. Сид фиксирует план прогона, так что упавшую итерацию можно повторить. В CI гоняется каждую ночь. И первый серьезный баг он нашел не в моем коде.
Находка № 1: read_committed, который читает отмененное
Время от времени Restore терял закоммиченные записи. Виноват был клиент: он хранил диапазоны отмененных транзакций по producer ID между fetch‑ответами. Когда продюсер с тем же ID после перезапуска отменял новую транзакцию, ее abort‑маркер растягивал старый диапазон, и под фильтр попадали закоммиченные записи. В одном прогоне их было 630.
Я описал баг в platformatic/kafka#467, его исправили в PR #470. Пока фикс не вышел в релиз, kstreams‑ts фильтрует fetch‑ответы сам, по алгоритму Java‑клиента.
Находка № 2: Kafka 3.9 и эпоха продюсера
На Kafka 4.4 все тесты зеленые, на 3.9 упали два, оба про таймаут транзакции.
Когда брокер отменяет транзакцию по таймауту, он поднимает эпоху продюсера. Worker потом снова занимает свой transactional ID через InitProducerId со старой эпохой. На 4.x это работает. А в 3.9 prepareFenceProducerEpoch еще и сбрасывает lastProducerEpoch, брокер отвечает PRODUCER_FENCED, и приложение останавливается как зомби.
Решил второй попыткой: если со старой эпохой не вышло, Worker пробует эпоху + 1. Брокер примет ее, только если после таймаута ничего не было. Если ID перехватил другой продюсер, эпоха выросла минимум на 2, и настоящий зомби все равно не пройдет.
Чего пока нет
Это 0.1, бета. Нет таблиц, join’ов, repartition‑топиков, hopping и session окон, standby‑реплик. Join потока с таблицей и группировку по другому ключу пока можно сделать руками, рецепты есть в документации. Нужны Node.js 24+ и Kafka 3.6+.
Попробовать
npm install kstreams-ts
-
Гарантии доставки и fencing: https://github.com/biogin/kstreams‑ts/blob/main/docs/delivery‑guarantees.md
Поэтому приглашаю всех попробовать. Жесткая критика приветствуется.
ссылка на оригинал статьи https://habr.com/ru/articles/1099332/