Синхронизация асинхронности: Dead Letter и Inbox для обработки зависимых сообщений

При проектировании интеграций через Kafka или другие брокеры сообщений можно столкнуться с проблемой зависимых сообщений в разных очередях/топиках.

Давайте изучим эту проблему подробнее и проанализируем, как ее можно решить с помощью паттернов проектирования Dead Letter и Inbox.

Проблема зависимых сообщений

Рассмотрим пример, когда надо через Kafka получать от другой системы несколько сущностей и сохранять их. Сущности примерно такие: продукты (Кефир Простоквашино 15%), продуктовые категории (Кефиры 15%), поставщики продуктов (ООО Простаквашино).

Логично, что каждая сущность передается в своем топике:

Продукты:

"product": {      "id": "1",      "name": "Кефир Простоквашино 15%",      "categoryId": "2",      "supplierId": "3"  }

Категории:

"category": {    "id": "2",      "name": "Кефиры 15%",      "parentCategoryId": "1"  }

Поставщики:

"supplier": {     "id": "3",     "name": "ООО Простаквашино",     "contractId": "1"  }

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

Например, создали нового поставщика «ООО Хлеб», и сразу же создали для него товар: «Хлеб белый». Но топики с продуктами и поставщиками могут быть подняты в разных инстансах. Или в топике с поставщиками большое количество сообщений, а в топике с продуктами пусто. В результате к нам приходит товар с supplierId, которого у нас еще нет.

То есть сначала приходит сообщение:

"product": {     "id": "2",     "name": "Хлеб белый",     "categoryId": "3",     "supplierId": "4"  }

И только через какой-то неопределенный промежуток времени приходит поставщик для этого продукта:

"supplier": {     "id": "4",     "name": "ООО Хлеб",     "contractId": "3"  }

Получается, что мы не можем сразу сохранить продукт, так как в таблице с продуктами поле с ID поставщика ссылается по Foreign Key на запись в таблице с поставщиками. Но в таблицу с поставщиками запись еще не поступила. А значит, БД нам не даст сохранить такой продукт.

Можно снять ограничение с поля поставщика в таблице с продуктами и сохранять продукт с каким-то несуществующим у нас ID поставщика. Но тогда данные в базе будут неконсистентными. То есть, будет непонятно:

• Что показывать пользователю в качестве поставщика этого продукта?

• Как создавать новые связи с этим поставщиком?

• Как делать расчеты на основе таких данных?

• Как поступать, если сообщение с поставщиком так и не придет?

И так далее, потенциальных проблем будет много.

В Data-продуктах вполне возможно закрыть подобные нестыковки через особенности проектирования хранилищ и получения оттуда данных, но в обычных проектах с реляционными БД это может быть сложно.

В общем, было бы хорошо уметь решать эту проблему еще на уровне интеграции, а не на уровне логики продукта.

Ситуация несколько напоминает race condition, но там потоки борются за обновление одного объекта. В нашем случае один поток должен дождаться другого, чтобы произвести обновление своего объекта.

Как же тогда обрабатывать зависимые сообщения?

Решение на уровне интеграции

Есть два паттерна проектирования, подходящие в том числе для решения подобных проблем: Dead letter и Inbox.

Оба этих подхода предполагают общий принцип решения проблемы зависимых сообщений:

1. Приходит сообщение с зависимыми данными, например, продукт с поставщиком, которого у нас еще нет в БД.

2. Данные из этого сообщения не сохраняются в таблицу с продуктами, вместо этого сообщение целиком сохраняется в специальное хранилище.

3. Ждем прихода сообщения с недостающим поставщиком, а когда его получаем — сохраняем поставщика в БД.

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

Как понять, когда сообщение с продуктом можно обрабатывать повторно?

Логику повторной обработки можно придумать любую, но чаще всего это просто polling и его вариации. То есть cron job с какой-то периодичностью пытается обработать каждое отложенное сообщение. Так происходит до тех пор, пока не случится одно из двух событий:

• Сообщение обработается, то есть в БД появится нужный поставщик и мы сможем сохранить продукт.

• Сообщение будет признано невалидным, то есть пройдет слишком много времени или количество попыток обработки превысит заданный лимит.

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

Несмотря на общий принцип решения проблемы, Dead Letter и Inbox существенно отличаются друг от друга. Эти различия касаются не только их целей, но и областей применения, а также методов реализации. Давайте рассмотрим каждый паттерн отдельно.

Dead Letter

При использовании Dead Letter паттерн создается отдельное хранилище (чаще всего это очередь или топик) для необработанных по любым причинам сообщений. Оно работает в режиме очереди и обычно называется Dead Letter Queue (DLQ). Сообщения из DLQ в последствии могут быть обработаны повторно или проанализированы вручную. Логика и частота повторной обработки задаются разработчиками, в зависимости от потребностей продукта.

Этот паттерн довольно прост в реализации и хорошо масштабируется. Многие брокеры сообщений (включая Kafka) имеют встроенную поддержку DLQ, что упрощает разработку и мониторинг. Также в DLQ естественным образом сохраняется порядок получения сообщений.

Большим преимуществом подхода является то, что проблемные сообщения выделяются в отдельную очередь и не мешают последовательной обработке сообщений из регулярного топика. То есть не создают дополнительной нагрузки на сервис и не тормозят работу обычных топиков.

Однако, у решения проблемы с зависимыми сообщениями с помощью Dead Letter есть ряд минусов:

• В одной очереди будут храниться и сообщения, для которых мы просто ожидаем недостающих данных, и проблемные сообщения, которые мы в принципе не сможем обработать. Это усложняет процесс для всех категорий отложенных сообщений.

• В данном случае для DLQ необходимо будет реализовывать довольно много специфичной логики.

• Если использовать одну DLQ на все топики, то в ней может скапливаться слишком много сообщений. При этом сообщения из разных топиков будут мешать обработке друг друга. Будет сложно расследовать и решать инциденты.

• Если создавать отдельную DLQ на каждый топик, что является стандартным решением, то на каждой из них придется делать отдельную логику обработки.

• Kafka и ее топики — это отдельный компонент инфраструктуры. Он может подвергаться проблемам доступности и требует специфических действий при реализации и обслуживании любых функций. То есть часто реализовать какую-то логику внутри сервиса дешевле и проще, чем реализовать ее на топике Kafka.

• В крупных компаниях SRE могут накладывать ограничения на функциональность Kafka. Например, период хранения сообщений (retention) может быть ограничен до 24 часов. Что неудобно, если мы собираемся вручную расследовать какие-то инциденты, а также если точно не знаем, с какой разницей во времени могут прийти зависимые сообщения.

Inbox

Inbox паттерн является общим подходом к обработке входящих сообщений. Он фокусируется на надежном хранении и последовательной обработке входящих событий, а не на изоляции проблемных сообщений.

При использовании этого подхода выделяется отдельное хранилище для всех входящих сообщений. Часто это таблица в БД, как и в Outbox.

Сначала все входящие сообщения сохраняются в таблицу Inbox, а потом берутся оттуда и обрабатываются в определенном порядке. Таким образом, процесс обработки отделен от процесса получения данных.

Так Inbox Pattern обеспечивает надежную доставку и обработку всех сообщений.

Также становится проще реализовывать специфичную логику обработки, например:

• Группировать сообщения по топикам и внутри топиков по агрегатам.

• Выделять топики в группы и для каждой группы задавать собственный порядок обработки.

• Добавить поле с количеством попыток обработки и задать приоритет обработки в зависимости от его значения.

• И так далее.

В общем, можно придумывать любые правила, на которые хватит компетенций и изобретательности.

Чтобы не обрабатывать бесконечно невалидные сообщения, можно создать для них отдельную таблицу (dead_letters). И переносить в эту таблицу записи, в некорректности которых мы уверены или у которых счетчик попыток достиг предельного значения.

Из минусов Inbox можно выделить следующие:

• Реализация может быть сложной, особенно при внедрении дополнительной логики определения порядка обработки и механизма перемещения некорректных сообщений в таблицу dead_letters.

• Требует выделения дополнительных ресурсов в БД и создает дополнительную нагрузку на базу.

• Может снизить скорость обработки сообщений, так как сначала сообщения сохраняются в таблицу, а потом в порядке очереди читаются из БД и обрабатываются.

• Если логика порядка обработки будет не оптимальной — это может спровоцировать накопление большого количества сообщений в Inbox.

• Все входящие сообщения сложены в одну “корзину”, то есть если с логикой или работой Inbox будут какие-то проблемы — пострадают все входящие потоки.

Dead Letter VS Inbox

Claude.ai сделала для нас сравнительную таблицу этих паттернов для решения задачи с зависимыми сообщениями:

Обобщая, можно сказать, что Inbox — более комплексное, но более надежное решение. Риск потерять сообщения и данные здесь будет ниже. Однако, стоит заложить дополнительные ресурсы на реализацию и последующую оптимизацию.

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

Заключение

Есть разные способы разобраться с зависимыми сообщениями. Dead Letter и Inbox — это общие паттерны, которые решают и другие задачи при обмене сообщениями. Поэтому полезно рассмотреть их среди прочих вариантов.

Итоговый выбор всегда зависит от конкретных требований вашего проекта, имеющихся ресурсов и специфики обрабатываемых данных.

А какие механизмы для решения подобных проблем использовали вы?

Другие мои материалы можно почитать в телеграм-канале breakfront.