❓ Что такое delivery semantics в Apache Kafka?
Delivery semantics [delivery guarantees, processing guarantees] — это способы доставки сообщений, т.е. как именно будут взаимодействовать между собой Producer, Broker и Consumer для того чтобы доставить сообщение.
Рассмотрим на примере классической схемы. У нас есть:
1⃣ Kafka Cluster с любым количеством Broker’ов
2⃣ Producer — читает данные из БД и отправляет их в виде сообщения Broker’у в demo.topic
3⃣ Consumer — подписан на demo.topic, если сообщение пришло, то читает его и сохраняет данные в БД
Для примера неважно какие используются БД у Consumer’a и Producer’a
Доставка сообщения в нашем случае может состоять из шагов:
Шаг 1. Producer читает данные из БД
Шаг 2. Producer отправляет сообщение Broker’y в demo.topic
Шаг 3. Broker сохраняет сообщение в demo.topic’e
Шаг 4. Broker уведомляет Producer’a, подтверждая что сообщение сохранено [acknowledgment]
Шаг 5. Consumer читает сообщение из demo.topic’a
Шаг 6. Consumer уведомляет Broker’a, что сообщение успешно прочитано [offset commit]
Шаг 7. Consumer сохраняет данные в БД
❓ Что может пойти не так?
На каждом шаге может произойти сбой, например:
❌ Закончилась оперативная память — OutOfMemoryError
❌ Закончилось место на диске
❌ Временный сбой в сети
❌ Случился timeout, например, Шаг 3 — Broker слишком долго сохранял сообщение
В зависимости от того, какие гарантии мы хотим получить в случае если что-то пойдет не так, мы используем разные delivery semantics.
Delivery semantics для Producer’a:
1⃣ At most once — сообщение будет отправлено максимум 1 раз. Producer отправляет сообщение и не ждет от Broker’a acknowledgment, а если Broker не смог его записать, сообщение больше отправляться не будет, оно утеряно. Принцип ‘fire-and-forget’.
2⃣ At least once — сообщение будет отправлено минимум 1 раз. Producer отправляет сообщение и ждет acknowledgment, а если Broker не отправил acknowledgment, то Producer заново отправит сообщение.
Тут может быть ситуация, если Broker записал сообщение, но вышел из строя на этапе отправки acknowledgment’a, и когда Broker восстановится и Producer переотправит сообщение, то мы по сути создадим дубликат, сохранив в Broker’e дважды одно сообщение.
3⃣ Exactly once — сообщение будет отправлено только 1 раз. Producer отравляет сообщение и ждет acknowledgment, а если Broker не отправил acknowledgment, то Producer заново отправит сообщение, и если Broker упадет на этапе отправки acknowledgment’a и получит дубликат, он это поймет и не будет записывать одно и тоже сообщение дважды. Здесь мы не теряем сообщения и не создаем дубликаты.
Delivery semantics для Consumer’a:
1⃣ At most once — сообщение будет прочитано максимум 1 раз. Consumer читает сообщение из Broker’a и если он не может его обработать, перечитывать его не будет.
У нас Шаг 7, например, Spring приложение Consumer упало с OOM и потом поднялось, но перечитывать сообщение не будет хоть и могло бы записать что-то в БД. Тут мы теряем обработку сообщения.
2⃣ At least once — сообщение будет прочитано минимум 1 раз. Consumer читает сообщение из Broker’a и если он не может его обработать, он заново его перечитает.
У нас Шаг 6 и Шаг 7 меняются местами, однако, например, мы делаем транзакцию в БД, хотим сделать commit offset и падаем с OOM, и когда поднимаемся, то перечитываем сообщение и делаем ту же транзакцию, что может создать дубликат в БД.
3⃣ Exactly once — сообщение будет прочитано только 1 раз. Consumer читает сообщение из Broker’a и если он не может его обработать, он заново его перечитает.
Чтобы не создать дубликат и не потерять обработку сообщения у нас есть 2 варианта:
— Транзакция в БД идемпотентна и если мы повторим ее, то ничего не изменится
— Мы делаем commit offset в рамках транзакции к БД [необязательно хранить offset в Broker’e, можно хранить их в БД], чтобы Шаг 6 и Шаг 7 был одним атомарным шагом. Тогда у нас нет дубликатов и обработку сообщения мы не теряем
Немного рефакторинга — 🐙Github
#sandbox_kafka #spring_boot_kafka
➿ Меню
➿ Подпишись: @developer_sandbox
Delivery semantics [delivery guarantees, processing guarantees] — это способы доставки сообщений, т.е. как именно будут взаимодействовать между собой Producer, Broker и Consumer для того чтобы доставить сообщение.
Рассмотрим на примере классической схемы. У нас есть:
1⃣ Kafka Cluster с любым количеством Broker’ов
2⃣ Producer — читает данные из БД и отправляет их в виде сообщения Broker’у в demo.topic
3⃣ Consumer — подписан на demo.topic, если сообщение пришло, то читает его и сохраняет данные в БД
Для примера неважно какие используются БД у Consumer’a и Producer’a
Доставка сообщения в нашем случае может состоять из шагов:
Шаг 1. Producer читает данные из БД
Шаг 2. Producer отправляет сообщение Broker’y в demo.topic
Шаг 3. Broker сохраняет сообщение в demo.topic’e
Шаг 4. Broker уведомляет Producer’a, подтверждая что сообщение сохранено [acknowledgment]
Шаг 5. Consumer читает сообщение из demo.topic’a
Шаг 6. Consumer уведомляет Broker’a, что сообщение успешно прочитано [offset commit]
Шаг 7. Consumer сохраняет данные в БД
❓ Что может пойти не так?
На каждом шаге может произойти сбой, например:
❌ Закончилась оперативная память — OutOfMemoryError
❌ Закончилось место на диске
❌ Временный сбой в сети
❌ Случился timeout, например, Шаг 3 — Broker слишком долго сохранял сообщение
В зависимости от того, какие гарантии мы хотим получить в случае если что-то пойдет не так, мы используем разные delivery semantics.
Delivery semantics для Producer’a:
1⃣ At most once — сообщение будет отправлено максимум 1 раз. Producer отправляет сообщение и не ждет от Broker’a acknowledgment, а если Broker не смог его записать, сообщение больше отправляться не будет, оно утеряно. Принцип ‘fire-and-forget’.
2⃣ At least once — сообщение будет отправлено минимум 1 раз. Producer отправляет сообщение и ждет acknowledgment, а если Broker не отправил acknowledgment, то Producer заново отправит сообщение.
Тут может быть ситуация, если Broker записал сообщение, но вышел из строя на этапе отправки acknowledgment’a, и когда Broker восстановится и Producer переотправит сообщение, то мы по сути создадим дубликат, сохранив в Broker’e дважды одно сообщение.
3⃣ Exactly once — сообщение будет отправлено только 1 раз. Producer отравляет сообщение и ждет acknowledgment, а если Broker не отправил acknowledgment, то Producer заново отправит сообщение, и если Broker упадет на этапе отправки acknowledgment’a и получит дубликат, он это поймет и не будет записывать одно и тоже сообщение дважды. Здесь мы не теряем сообщения и не создаем дубликаты.
Delivery semantics для Consumer’a:
1⃣ At most once — сообщение будет прочитано максимум 1 раз. Consumer читает сообщение из Broker’a и если он не может его обработать, перечитывать его не будет.
У нас Шаг 7, например, Spring приложение Consumer упало с OOM и потом поднялось, но перечитывать сообщение не будет хоть и могло бы записать что-то в БД. Тут мы теряем обработку сообщения.
2⃣ At least once — сообщение будет прочитано минимум 1 раз. Consumer читает сообщение из Broker’a и если он не может его обработать, он заново его перечитает.
У нас Шаг 6 и Шаг 7 меняются местами, однако, например, мы делаем транзакцию в БД, хотим сделать commit offset и падаем с OOM, и когда поднимаемся, то перечитываем сообщение и делаем ту же транзакцию, что может создать дубликат в БД.
3⃣ Exactly once — сообщение будет прочитано только 1 раз. Consumer читает сообщение из Broker’a и если он не может его обработать, он заново его перечитает.
Чтобы не создать дубликат и не потерять обработку сообщения у нас есть 2 варианта:
— Транзакция в БД идемпотентна и если мы повторим ее, то ничего не изменится
— Мы делаем commit offset в рамках транзакции к БД [необязательно хранить offset в Broker’e, можно хранить их в БД], чтобы Шаг 6 и Шаг 7 был одним атомарным шагом. Тогда у нас нет дубликатов и обработку сообщения мы не теряем
Немного рефакторинга — 🐙Github
#sandbox_kafka #spring_boot_kafka
➿ Меню
➿ Подпишись: @developer_sandbox