Kafka Streams для Node.js: как я делал exactly‑once на TypeScript и что сломалось по дороге
Сразу: проект сделан с помощью ИИ. Если вы против, статью можно пропустить.
Сам не люблю иишный слоп. Берусь только за то, что мне реально мешало/мешает, и все проверяю сам. Хотел попробовать 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()sourcecraft.dev
Типы проходят через всю цепочку: если тип ключа или значения на выходе не тот, код не скомпилируется. Кроме 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-tssourcecraft.dev
GitHub: biogin/kstreams-ts
Гарантии доставки и fencing: github.com/biogin/kstreams-ts/blob/main/docs/deli…
Поэтому приглашаю всех попробовать. Жесткая критика приветствуется.
Mentions
不改代码，Node.js 堆内存直接砍半？ Platformatic 开源了 node-caged：开启 V8 指针压缩，Docker 镜像一换即可。 实测对象数组等结构内存≈50%↓ 限制：每 Isolate 4GB（Worker 可叠加） GitHub：platformatic/node-caged docker pull platformatic/node-caged:26-slim 你们容器一般给多少内存？
目前项目已经用 GCC 14 预编译好了 Node.js 26.x 的多架构镜像（支持 Debian trixie / slim / alpine，同时内置了 Temporal 支持），一条 `docker pull platformatic/node-caged:26-slim` 就能直接替换现有容器底座。 GitHub 仓库： platformatic/node-caged Docker Hub 镜像： hub.docker.com 你们现在的 Node.js 容器单实例一般给分配多少内存？有超过 4GB 的场景吗？
Kafka Streams для Node.js: как я делал exactly‑once на TypeScript и что сломалось по дороге Сразу: проект сделан с помощью ИИ. Если вы против, статью можно пропустить. Сам не люблю иишный слоп. Берусь только за то, что мне реально мешало/мешает, и все проверяю сам. Хотел попробовать 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()sourcecraft.dev Типы проходят через всю цепочку: если тип ключа или значения на выходе не тот, код не скомпилируется. Кроме 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-tssourcecraft.dev GitHub: biogin/kstreams-ts Гарантии доставки и fencing: github.com/biogin/kstreams-ts/blob/main/docs/deli… Поэтому приглашаю всех попробовать. Жесткая критика приветствуется.
@matteocollina @napi @platformatic I'd like to use it as a drop-in replacement - but we require using the node user - do you think you would add it? node-caged#1
@napi @platformatic If your app is RAM-bound and tight container limits, serverless cold starts, and high memory bills are hitting you, this is the easiest 50% memory win you'll ever test. Two minutes with Docker. Try it: platformatic/node-caged
If your app is RAM-bound and tight container limits, serverless cold starts, and high memory bills are hitting you, this is the easiest 50% memory win you'll ever test. Two minutes with Docker. Try it: platformatic/node-caged
Node has a bright future. See e.g. Platformatic's "Watt" server^1 for an example of pushing it forward. 1. platformatic.dev and github.com/platformatic
@thdxr very interesting approach! Take a look at what can be done with latest @nodejs. platformatic/secure-eval-worker