Обычный слушатель читает запись и делает по ней одно действие: списал деньги, отправил письмо. Часть задач в такую модель не помещается: посчитать заказы каждого магазина за последние пять минут, склеить платёж с заказом, который пришёл в другой топик, держать актуальный адрес покупателя из потока правок профиля. Каждой из них нужна память между записями — счётчик, буфер ожидания, последняя версия. Написать такую память поверх consumer можно, но тогда придётся самому решать, где она живёт, что с ней станет при падении экземпляра и как поделить её между двумя экземплярами. Это и есть потоковая обработка, и Kafka Streams делает её обычной Java-библиотекой: без отдельного кластера, внутри вашего сервиса.
Разница с обычным consumer — в состоянии: счётчики и окна лежат на диске экземпляра, а копия каждого изменения уходит в Kafka, и из неё состояние собирается заново после падения.
Счётчики живут в локальном RocksDB экземпляра, но каждое их изменение дублируется в changelog-топик. Упал экземпляр — новый вычитал changelog, восстановил shop-1 → 3 и shop-2 → 1 и продолжил счёт с них, а не с нуля.
Как это запускается в Spring Boot
Отдельного кластера нет, но у приложения появляются своя группа потребителей, служебные топики и каталог на диске — и всё это привязано к одному имени.
spring-kafka подключает Streams аннотацией @EnableKafkaStreams: Spring Boot собирает из spring.kafka.streams.* бин KafkaStreamsConfiguration, а StreamsBuilderFactoryBean строит топологию из ваших бинов, запускает KafkaStreams при старте контекста и закрывает при остановке. Единственная обязательная настройка — application.id (без неё Boot возьмёт spring.application.name, а без обоих не стартует). Это не просто имя: оно становится group.id для чтения входных топиков, префиксом всех служебных топиков (<application.id>-...-changelog, -repartition) и именем каталога состояния на диске. Смените его — и приложение начнёт с чистого листа: новая группа читает топики с начала или с конца по auto.offset.reset, а старые changelog-топики останутся мусором в кластере.
spring:
kafka:
bootstrap-servers: kafka-1:9092,kafka-2:9092,kafka-3:9092
streams:
application-id: order-stats
state-dir: /var/lib/order-stats/state
@Configuration
@EnableKafkaStreams
public class OrderStatsTopology {
@Bean
public KStream<String, Order> ordersPerShop(StreamsBuilder builder, JsonSerde<Order> orderSerde) {
KStream<String, Order> orders = builder.stream("orders", Consumed.with(Serdes.String(), orderSerde));
orders.filter((shopId, order) -> order.amount().signum() > 0)
.groupByKey()
.count(Materialized.as("orders-per-shop"))
.toStream()
.to("orders-per-shop", Produced.with(Serdes.String(), Serdes.Long()));
return orders;
}
}
Возвращаемый KStream Spring никак не использует: важно, что метод зависит от StreamsBuilder, поэтому топология описана до того, как StreamsBuilderFactoryBean её запустит. Materialized.as("orders-per-shop") даёт хранилищу имя — оно станет частью имени changelog-топика и понадобится, когда счётчик захочется прочитать снаружи. После count() значением становится Long, а не исходный заказ, поэтому и сериализатор на выходе нужен другой: забудете Produced.with(...) — топология возьмёт сериализатор по умолчанию из конфигурации и упадёт уже в рантайме на приведении типов. Первое, что стоит вывести в лог при старте, — streamsBuilderFactoryBean.getTopology().describe(): он печатает подтопологии, хранилища и все служебные топики, которые приложение создаст в кластере.
KStream и KTable: тот же топик, два прочтения
В топике customers лежит история правок профиля: покупатель сменил город, потом телефон, потом снова город. Чтобы подставить в заказ актуальный адрес, нужна не история, а последняя версия по каждому покупателю — и хранить её должен кто-то, кто переживёт перезапуск сервиса.
Один и тот же топик Kafka Streams читает двумя способами. builder.stream("customers") даёт KStream: каждая запись — самостоятельный факт, третья правка ничего не делает с первой, обработка видит все пять. builder.table("customers") даёт KTable: записи с одним ключом перезаписывают друг друга, таблица держит последнее значение по ключу в state store, а запись с null вместо значения удаляет строку. Пять фактов о трёх покупателях становятся тремя строками.
Один топик, два прочтения. KStream копит факты: пять записей — пять строк. KTable держит последнее значение по ключу: две правки u-1 (выделены) перезаписывают одну и ту же строку, и из пяти записей остаются три.
KTable<String, Customer> customers = builder.table("customers",
Consumed.with(Serdes.String(), customerSerde),
Materialized.as("customers-store"));
Топику под KTable ставят cleanup.policy=compact: брокер оставляет последнюю запись по каждому ключу, и при старте таблица вычитывается по объёму текущих строк, а не всей истории. Результат count() из предыдущего раздела — тоже KTable: каждая новая запись обновляет счётчик, а toStream() превращает обновления в поток «ключ → новое значение». Поэтому в выходном топике для одного магазина лежит много записей, а не одна, и читатель обязан брать последнюю по ключу — это changelog счётчика, а не его срез.
Есть и третий вариант — GlobalKTable: каждый экземпляр держит полную копию таблицы, а не только свои партиции. Так читают справочники — тарифы, курсы валют, тысячи строк, а не миллионы: копия лежит на каждом узле, зато join с ней не требует общего ключа, о чём ниже.
Ключ решает всё: партиции и скрытые repartition-топики
Заказы в orders лежат с ключом магазина, а отчёт нужен по регионам. Меняете ключ — и в кластере появляется топик, который вы не создавали.
Streams делит работу на задачи по партициям входного топика: задача 3 обрабатывает партицию 3 у каждого своего входного топика и держит состояние только по ключам, которые в неё попадают. Пока ключ не меняется, все заказы одного магазина лежат в одной задаче, и счётчик по магазину честный. Но groupBy((shop, order) -> order.region()) меняет ключ, и заказы одного региона теперь разбросаны по всем партициям, то есть по всем задачам. Чтобы собрать их вместе, Streams пишет поток обратно в Kafka — в служебный топик <application.id>-<имя>-repartition — и читает его заново, уже разложенным по новому ключу. Каждая запись проходит через брокер дважды: лишний сетевой трафик, лишний диск на брокере и задержка на запись — от десятков миллисекунд до интервала коммита.
selectKey, map, flatMap и process помечают поток как «ключ мог измениться»; сама перекладка случится на следующей операции с состоянием — groupByKey, aggregate, join. groupBy с функцией ключа перекладывает всегда. Лечится это выбором ключа на стороне продьюсера: если отчёты по регионам нужны постоянно, пусть заказы изначально пишутся с ключом региона или в отдельный топик. Явный repartition(Repartitioned.numberOfPartitions(12)) нужен, когда перекладка неизбежна и ей хочется задать имя и число партиций, а не полагаться на сгенерированные.
Смена ключа разрывает конвейер надвое: записи уходят в служебный топик и читаются заново, уже разложенные по региону.
Окна: время события, опоздавшие записи и grace
«Заказов за пять минут» — задача звучит просто, пока не спросить, за какие пять минут: по часам сервера в момент обработки или по времени, когда заказ создан. Мобильное приложение отправляет события пачкой, когда появится сеть, и заказ, сделанный в 12:04, приезжает в 12:09. По часам обработки он попадёт в чужое окно; по времени события — в своё, но окно 12:00–12:05 к этому моменту уже могло закрыться.
Kafka Streams по умолчанию считает по времени события: берёт timestamp записи, который продьюсер ставит при отправке (CreateTime) либо брокер при записи (LogAppendTime). Если время лежит внутри самого события, его достаёт свой TimestampExtractor; для «по часам обработки» есть готовый WallclockTimestampExtractor. Время в задаче не идёт по часам — оно продвигается максимальным timestamp среди увиденных записей, и это называется stream time. Окно закрывается не тогда, когда на часах 12:05, а когда пришла запись с временем позже 12:05 плюс grace.
Grace — сколько окно ждёт опоздавших после своей границы. Запись, чьё время относится к окну 12:00–12:05, а stream time уже дальше 12:05 плюс grace, молча отбрасывается: ни исключения, ни записи в DLT, только рост метрики dropped-records-total у задачи. До версии 3.0 grace по умолчанию был 24 часа, и приложения «не выдавали результат», потому что ждали сутки; с 3.0 его задают явно — ofSizeAndGrace или ofSizeWithNoGrace.
Сверху вниз порядок прихода: третья запись двигает stream time за границу окна с grace, поэтому последняя, с временем 12:01:40, уже не попадает в закрытое окно.
static void ordersPerFiveMinutes(StreamsBuilder builder, Serde<Order> orderSerde, Serde<ShopWindowCount> countSerde) {
builder.stream("orders", Consumed.with(Serdes.String(), orderSerde))
.groupByKey()
.windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(2)))
.count(Materialized.as("orders-per-5min"))
.suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()))
.toStream()
.map((shop, count) -> KeyValue.pair(shop.key(), new ShopWindowCount(shop.window().startTime(), count)))
.to("orders-per-5min", Produced.with(Serdes.String(), countSerde));
}
Без suppress окно выдаёт промежуточные результаты — 1, 2, 3 на каждое событие, с учётом кеша пачками раз в commit.interval.ms, по умолчанию 30 секунд. Для панели, где счётчик растёт на глазах, это то, что нужно; для алерта «больше 100 заказов за 5 минут» — сотня лишних алертов. suppress(untilWindowCloses) держит результат до закрытия окна и выдаёт одну финальную запись. Цена: результат появляется на grace позже конца окна, а буфер ждущих окон живёт в памяти по ключам с собственным changelog. unbounded() значит «сколько понадобится», и на миллионе активных ключей это гигабайты; maxRecords(n).shutDownWhenFull() остановит приложение раньше, чем оно съест всю память.
Кроме окон встык есть скользящие (advanceBy: размер 5 минут, шаг 1 минута — каждое событие попадает в пять окон, состояния впятеро больше), сессионные (SessionWindows — окно закрывается по паузе между событиями ключа) и SlidingWindows для «за последние N минут от каждого события». Хранилище окна держит их размер плюс grace; changelog у оконного хранилища — с политикой delete,compact и сроком на сутки больше, чтобы восстанавливающийся экземпляр успел дочитать.
Join двух потоков: пара, окно и ко-партиционирование
Платёж приходит в payments, заказ — в orders, и одно без другого бесполезно: «оплаченный заказ» надо собрать из двух записей, которые приезжают с разницей в секунды или минуты и в любом порядке. На стенде join работает; в production половина платежей остаётся без пары, а в логах пусто.
KStream<String, Order> orders = builder.stream("orders", Consumed.with(Serdes.String(), orderSerde));
KStream<String, Payment> payments = builder.stream("payments", Consumed.with(Serdes.String(), paymentSerde));
orders.join(payments,
(order, payment) -> new PaidOrder(order, payment),
JoinWindows.ofTimeDifferenceAndGrace(Duration.ofMinutes(10), Duration.ofMinutes(1)),
StreamJoined.with(Serdes.String(), orderSerde, paymentSerde))
.to("paid-orders", Produced.with(Serdes.String(), paidOrderSerde));
Обе стороны попадают в оконные хранилища: заказ ждёт платёж, платёж ждёт заказ — по 10 минут в каждую сторону, потому что окно симметричное. Пришла пара в пределах окна — на выход уходит PaidOrder; платёж через 16 минут после заказа пары не получит, и это не ошибка, а результат по определению. leftJoin выдаст заказ с null вместо платежа, когда окно закроется — с версии 3.1 именно после закрытия, а не сразу, чтобы не плодить ложных «неоплаченных».
Каждая сторона join ждёт вторую в оконном хранилище. Платёж o-7 пришёл через 6 минут после заказа и попал в окно — пара ушла в paid-orders. Платёж o-8 пришёл через 16 минут: окно закрылось, пары нет, и никакой ошибки Streams не покажет. Окно симметричное — платёж, пришедший раньше заказа, точно так же дождётся его.
Join сравнивает ключи внутри одной задачи, а задача видит по одной партиции каждого топика: партицию 3 в orders и партицию 3 в payments. Пара найдётся, только если заказ и его платёж легли в партиции с одинаковым номером. Для этого у топиков должно совпадать число партиций и способ раскладки ключей по ним — это называется ко-партиционирование. Число партиций Streams проверяет сам: при распределении задач он сравнит orders с 12 партициями и payments с 6 и упадёт с TopologyException — заметно и сразу.
А вот способ раскладки проверить нельзя: если платежи пишет сервис на другом языке со своим партиционером или продьюсер заказов кладёт ключ строкой "42", а платежей — числом 42, номера партиций разойдутся, и часть пар молча не найдётся. Лечение — переложить меньшую сторону явно: payments.repartition(Repartitioned.numberOfPartitions(12)) с тем же партиционером, что у большей. Исключения два: GlobalKTable (полная копия на каждом экземпляре, ключ подбирается функцией) и join двух KTable по внешнему ключу, где Streams перекладывает данные сам.
Join потока с таблицей устроен иначе: это не ожидание пары, а поиск по ключу в момент прихода записи потока. Заказ пришёл — Streams заглянул в customers и подставил адрес. Если правка профиля приедет секундой позже заказа, заказ уйдёт со старым адресом: таблица не ждёт, окна у такого join нет. Ко-партиционирование нужно и здесь.
Что случится при падении: changelog, standby и диск
Экземпляр упал, его партиции с их задачами переехали к соседу. У обычного consumer сосед продолжает с закоммиченного offset за секунды. У Streams сосед должен сначала получить состояние упавшего — а оно лежало на чужом диске.
Каждый state store по умолчанию дублируется в changelog-топик: каждое изменение счётчика — запись ключ → значение, топик с компакцией, по каждому ключу остаётся последнее. Новый владелец задачи создаёт пустой RocksDB и вычитывает changelog с начала — это восстановление, и всё это время задача не обрабатывает входные записи. Скорость — как у обычного consumer, десятки мегабайт в секунду: состояние в 10 ГБ — это минуты простоя, и они повторяются при каждом деплое, потому что деплой — тоже переезд задач. Changelog можно выключить (withLoggingDisabled()), но тогда падение экземпляра теряет состояние насовсем: так делают только для хранилищ, которые пересчитываются из входного топика.
Паузу сокращают с двух сторон. Первая — не терять диск: state.dir на томе, который переживает перезапуск (в Kubernetes — StatefulSet с PersistentVolume, а не Deployment с пустым emptyDir). При старте Streams сверяет файл .checkpoint в каталоге задачи с changelog и дочитывает только хвост.
Вторая — резервные копии: num.standby.replicas=1 заставляет другой экземпляр держать копию состояния каждой задачи, читая её changelog в фоне. Упал активный — задача уходит туда, где копия уже лежит, и дочитать надо только отставание последних секунд. Цена — двойной диск под состояние и двойное чтение changelog. Планировщик задач с версии 2.6 смотрит на acceptable.recovery.lag, по умолчанию 10 000 записей: экземпляр, чья копия отстаёт меньше, считается готовым и получает задачу сразу; иначе задача остаётся у готового, на новом экземпляре копия «разогревается» в фоне, и переезд случается на пробном ребалансе — раз в probing.rebalance.interval.ms, по умолчанию 10 минут.
Когда Kafka Streams, а когда нет
Берите, когда нужны агрегации по времени, join потоков или поддержание производного состояния — и данные уже лежат в Kafka с обеих сторон.
Не берите, если обработка — «прочитал запись, вызвал сервис, записал в базу»: состояния нет, окон нет, а Streams добавит changelog-топики, ребаланс задач и каталог на диске без всякой пользы. Обычный @KafkaListener с идемпотентностью здесь честнее. Не берите и когда данных нет в Kafka: Streams читает и пишет только топики; файлы, базы и чужие очереди подключают через Kafka Connect по бокам — или это уже не задача Streams.
Flink вместо Streams — по четырём признакам. Параллелизм нужен больше числа партиций: Streams не даст задач больше, чем партиций во входном топике, а Flink делит работу по своим слотам независимо от источника. Состояние измеряется сотнями гигабайт: восстановление из changelog займёт часы, а Flink снимает инкрементальные снимки в объектное хранилище. Нужен SQL по потокам или поиск цепочек событий (CEP). Или у вас уже есть команда, которая эксплуатирует кластер Flink. За всё это платят второй системой рядом с Kafka: JobManager, TaskManager, их обновление и мониторинг.
Глубже: спросить состояние снаружи: интерактивные запросырасширенное
Счётчик заказов по магазину уже лежит в state store. Отдавать его в API через ещё один топик и базу — лишний крюк: хранилище можно опросить прямо из сервиса.
@RestController
@RequiredArgsConstructor
public class ShopStatsController {
private final KafkaStreamsInteractiveQueryService queries;
@GetMapping("/shops/{shopId}/orders-count")
public long ordersCount(@PathVariable String shopId) {
ReadOnlyKeyValueStore<String, Long> store =
queries.retrieveQueryableStore("orders-per-shop", QueryableStoreTypes.keyValueStore());
Long count = store.get(shopId);
return count == null ? 0 : count;
}
}
KafkaStreamsInteractiveQueryService появился в Spring Kafka 3.2; раньше хранилище брали напрямую — streamsBuilderFactoryBean.getKafkaStreams().store(StoreQueryParameters.fromNameAndType(...)). Два подводных камня. Первый — состояние поделено между экземплярами: на этом узле лежат счётчики только его партиций, и запрос по чужому магазину вернёт null, хотя заказы есть. Поэтому в конфигурации указывают application.server=host:port своего экземпляра, перед чтением спрашивают, где лежит ключ (getKafkaStreamsApplicationHostInfo, под капотом queryMetadataForKey), и перенаправляют HTTP-запрос на нужный экземпляр — этот прокси пишете вы.
Второй — во время ребаланса и восстановления хранилище недоступно: store() бросает InvalidStateStoreException, запрос стоит повторить через секунду (сервис Spring делает это сам через RetryTemplate); если устаревший ответ лучше отказа, StoreQueryParameters#enableStaleStores() разрешает читать резервную копию.
Глубже: exactly-once внутри Kafkaрасширенное
Конвейер «прочитал из orders, посчитал, записал в orders-per-shop» может упасть между отправкой результата и коммитом offset — после перезапуска результат уйдёт второй раз, и счётчик задвоится. Режим processing.guarantee=exactly_once_v2 объединяет чтение, запись состояния в changelog, отправку результата и сдвиг offset в одну транзакцию Kafka: либо всё зафиксировано, либо ничего. Для конвейеров «топик → обработка → топик» это честный exactly-once без самодельной идемпотентности; старые режимы exactly_once и exactly_once_beta удалены в 4.0. Граница гарантии — стены Kafka: вызов внешнего REST API или запись в базу внутри топологии транзакцией не покрываются, там идемпотентность остаётся вашей заботой.
Две вещи стоит знать до включения. Транзакциям нужен служебный топик __transaction_state, и по умолчанию он требует трёх реплик: на кластере из трёх брокеров режим включится, на одном брокере приложение упадёт при старте, пока не понизите настройки транзакционного лога. И результат становится виден читателям с read_committed только по фиксации транзакции — поэтому с exactly_once_v2 commit.interval.ms сам меняется с 30 секунд на 100 миллисекунд: иначе задержка каждой записи росла бы до полуминуты.
Глубже: тест топологии без брокерарасширенное
Проверять окна на живом кластере — значит ждать пять минут на каждый прогон и получать разный результат в зависимости от того, в какую секунду тест стартовал. Топология — функция от входных записей и их времени, и для неё есть драйвер, который прогоняет записи без брокера.
class OrdersPerFiveMinutesTest {
private TopologyTestDriver driver;
private TestInputTopic<String, Order> orders;
private TestOutputTopic<String, ShopWindowCount> counts;
@BeforeEach
void setUp() {
StreamsBuilder builder = new StreamsBuilder();
OrderStatsTopology.ordersPerFiveMinutes(builder, orderSerde, countSerde);
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-stats-test");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
driver = new TopologyTestDriver(builder.build(), props);
orders = driver.createInputTopic("orders", new StringSerializer(), orderSerde.serializer());
counts = driver.createOutputTopic("orders-per-5min", new StringDeserializer(), countSerde.deserializer());
}
@AfterEach
void tearDown() {
driver.close();
}
@Test
void lateOrderIsDroppedAfterGrace() {
Instant windowStart = Instant.parse("2026-03-01T12:00:00Z");
orders.pipeInput("shop-1", order(100), windowStart.plusSeconds(30));
orders.pipeInput("shop-1", order(200), windowStart.plusSeconds(90));
orders.pipeInput("shop-2", order(50), windowStart.plus(Duration.ofMinutes(7)).plusSeconds(1));
orders.pipeInput("shop-1", order(300), windowStart.plusSeconds(100));
assertThat(counts.readKeyValuesToList())
.containsExactly(KeyValue.pair("shop-1", new ShopWindowCount(windowStart, 2L)));
}
}
Драйвер обрабатывает каждую pipeInput синхронно, а время события задаётся третьим аргументом — и именно оно двигает stream time. Третья запись переводит его за 12:07, окно 12:00–12:05 с grace 2 минуты закрывается, suppress выдаёт финальный счётчик 2; четвёртая запись с временем 12:01:40 относится к закрытому окну и отбрасывается. Состояние тоже доступно — driver.getKeyValueStore("orders-per-shop") — для проверки таблиц без выходного топика. Зависимость — org.apache.kafka:kafka-streams-test-utils.
Чего драйвер не проверяет — ко-партиционирование: у него по одной партиции на топик, и join, который в production молча теряет пары, в тесте найдёт все. Числа партиций и партиционеры проверяет только тест на контейнере, как в Kafka в production.
Коротко
application.id— сразу group.id, префикс служебных топиков и каталог состояния; сменить его значит начать с чистого листа.- KStream копит факты, KTable держит последнее значение по ключу; результат
count()— KTable, и в выходном топике лежит его changelog, а не срез. - Смена ключа порождает repartition-топик, и каждая запись проходит через брокер дважды; ключ выбирают у продьюсера.
- Окна считают по времени события и закрывают по stream time; опоздавшие после grace отбрасываются молча (
dropped-records-total), один результат на окно даётsuppressценой задержки. - Join ждёт пару в симметричном окне; разное число партиций Streams поймает при старте, разный партиционер — нет, и пары молча теряются.
- Восстановление читает changelog целиком: 10 ГБ — минуты простоя на каждый деплой; лечат диск, переживающий рестарт, и
num.standby.replicas=1. - Интерактивные запросы видят только свои партиции; за чужим ключом идут на другой экземпляр, во время ребаланса хранилище недоступно.
exactly_once_v2закрывает дубли внутри Kafka и снижаетcommit.interval.msдо 100 мс; база и HTTP — по-прежнему идемпотентность.TopologyTestDriverпроверяет окна и grace без брокера, но не ловит ошибок ко-партиционирования.- Без состояния — обычный слушатель; данные не в Kafka, параллелизм выше числа партиций, состояние в сотни гигабайт, SQL по потокам — Flink.
Что почитать дальше
- Основы Kafka — партиции, ключ и consumer groups: почему у Streams не бывает задач больше, чем партиций.
- Kafka в production — транзакции Kafka, идемпотентный потребитель и тест на контейнере: то, что закрывает границы exactly-once.
- Event Sourcing — KTable как материализованный «текущий вид» из журнала событий.
- Синхронно или асинхронно — когда события вообще нужны, а когда хватит вызова.