Потоки Kafka фиксируют семантику смещения

Question

Я просто хотел подтвердить кое-что, что, как мне кажется, находится между строкой документации. Было бы правильно сказать, что Commit в потоках kafka не зависит от того, обработано ли offset / message всем набором узлов обработки топологии приложения, но зависит исключительно от интервал фиксации? Другими словами, там, где в типичном приложении-потребителе kafka, можно зафиксировать, когда сообщение полностью обработано, а не только выборка, в потоке Kafka простой выборки достаточно для интервала фиксации , чтобы сработать и зафиксировать что сообщение / смещение? То есть, даже если это смещение / сообщение еще не было обработано всем набором узлов обработки топологии приложения?

Или сообщение может быть зафиксировано, исходя из факта что весь набор узлов обработки топологии обработал их, и они готовы к go выходу в топи c или внешней системе.

В некотором смысле вопрос можно резюмировать как, когда есть ли смещение / сообщения, которые могут быть зафиксированы в потоках Kafka? это условно? если да, то в каком состоянии?

Answer 1

Независимо от того, используются ли потоки или просто потребитель, ключевым моментом является то, что автоматическая фиксация происходит в потоке опроса, а не в отдельном потоке - смещение пакета сообщений фиксируется только при последующем опросе, а commit.interval.ms просто определяет минимальное время между коммитами, ie большое значение означает, что фиксация не будет происходить при каждом опросе.

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

Answer 2

Вы должны понимать, что программа Kafka Streams, то есть ее Topology my, содержит несколько суб-топологий (https://docs.confluent.io/current/streams/architecture.html#stream -partitions-and-tasks ). Субтопологии связаны друг с другом через темы.

Запись может быть зафиксирована, если она полностью обрабатывается субтопологией. В этом случае промежуточный вывод записи записывается в topi c, который соединяет две суб-топологии до того, как произойдет фиксация. Подтопология нисходящего потока будет считывать данные из «подключения topi c» и фиксировать смещения для этой topi c.

На самом деле фиксация происходит только на основе commit.interval.ms. Если выборка возвращает, допустим, 100 записей (смещения от 0 до 99), и 30 записей обрабатываются субтопологией при попадании commit.interval.ms, Kafka Streams сначала проверяет, что вывод этих 30 сообщений сбрасывается в Kafka (ie, Producer.flush()), а затем будет зафиксировано смещение 30 - остальные 70 сообщений находятся только во внутреннем буфере потоков Kafka и будут обработаны после фиксации. Если буфер пуст, будет отправлена ​​новая выборка. Каждый поток отслеживает commit.interval.ms независимо и фиксирует все свои задачи, если интервал фиксации пройден.

Поскольку фиксация происходит на основе субтопологии, это может быть чем фиксируется входная запись topi c, в то время как выходной topi c еще не имеет данных результатов, потому что промежуточные результаты еще не обрабатываются подчиненной топологией.

Вы можете проверить структуру своей программы через Topology#describe(), чтобы увидеть какие суб-топологии есть в вашей программе.