«Точно один раз» только для потоков (topic1 -> app -> topic2)?

Question

У меня есть архитектура, в которой у нас есть два отдельных приложения. Первоначальный источник - база данных sql. Приложение 1 прослушивает таблицы CDC для отслеживания изменений в таблицах в этой базе данных, нормализует и сериализует эти изменения. Он берет эти сериализованные сообщения и отправляет их в тему Кафки. App2 прослушивает эту тему, адаптирует сообщения к различным форматам и отправляет эти адаптированные сообщения по соответствующим адресатам через HTTP.

Итак, наша потоковая архитектура выглядит так:

SQL (событие CDC) -> App1 (нормализует события) -> Kafka -> App2 (адаптирует события к конечным точкам) -> различные конечные точки

Мы надеемся добавить обработку ошибок в случае сбоя и не можем допускать повторяющиеся события, пропущенные события или изменение порядка. Учитывая вышеприведенную архитектуру, все, что нас действительно волнует, это то, что один раз относится к сообщениям, поступающим из App1 в App2 (наши отдельные производители и потребители)

Все, что я читаю, и каждый найденный мной пример транзакционного API указывает на «потоковую передачу». Похоже, что потоковый API Kafka предназначен для отдельного приложения, которое принимает входные данные из темы Kafka, выполняет их обработку и выводит их в другую тему Kafka, которая, по-видимому, не относится к нашему использованию Kafka. Вот выдержка из Документов Конфлюента :

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

Я изо всех сил пытаюсь обернуть голову, как мы можем использовать точно один раз с нашей темой Kafka, или если точно один раз Kafka создан даже для не «потоковых» сценариев использования. Придется ли нам строить собственную дедупликацию и отказоустойчивость?

Answer 1

Если вы используете Kafka Streams API (или другой инструмент, поддерживающий однократную обработку с помощью Kafka), то семантика (EOS) Kafka распространяется на все приложения:

topic A --> App 1 --> topic B --> App 2 --> topic C

В вашем случае использования один вопрос заключается в том, поддерживает ли начальный этап CDC EOS. Другими словами, вы должны задать вопрос: какие шаги включены, и все ли шаги покрываются EOS?

В следующем примере EOS поддерживается сквозной, если (и только если) начальный шаг CDC также поддерживает EOS, как и остальная часть потока данных.

SQL --CDC--> topic A --> App 1 --> topic B --> App 2 --> topic C

Если вы используете Kafka Connect для этапа CDC, вы должны проверить, поддерживает ли используемый вами разъем EOS, да или нет.

Все, что я читаю, и каждый найденный мной пример транзакционного API указывает на «потоковую передачу».

Транзакционный API производителей / потребителей Kafka предоставляет примитивы для обработки EOS. Kafka Streams, который расположен поверх клиентов производителя / потребителя, использует эту функциональность для реализации EOS таким образом, чтобы его можно было легко использовать разработчикам с несколькими строками кода (например, автоматически позаботиться об управлении состоянием, когда требуется приложение). выполнить операцию с состоянием, такую ​​как объединение или объединение). Возможно, эта связь между производителем / потребителем <-> Kafka Streams была вашей путаницей после прочтения документации?

Конечно, вы также можете «создавать свои собственные», используя базовых производителей и клиентов-клиентов Kafka (с транзакционными API) при разработке ваших приложений, но это больше работы.

Я изо всех сил пытаюсь осмыслить, как мы можем использовать точно один раз с нашей темой Kafka, или если точно один раз Kafka создан даже для не «потоковых» сценариев использования. Придется ли нам строить собственную дедупликацию и отказоустойчивость?

Не уверен, что вы подразумеваете под "не потоковыми" вариантами использования. Если вы имеете в виду: «Если мы не хотим использовать Kafka Streams или KSQL (или другой существующий инструмент, который может читать из Kafka для обработки данных), что нам нужно для достижения EOS в наших приложениях?», Тогда ответ будет следующим: «Да, в этом случае вы должны напрямую использовать производителя / клиентов Kafka и убедиться, что все, что вы делаете с ними, правильно реализует обработку EOS». (И поскольку последнее сложно, эта функциональность EOS была добавлена ​​в Kafka Streams.)

Надеюсь, это поможет.