Confluent Kafka — производитель InitTransaction приводит к исключению KafkaException ⇐ C#

[Anonymous]

Мы уже несколько месяцев внедряем Confluent Kafka, в частности, используя библиотеку C# https://github.com/confluentinc/confluent-kafka-dotnet. Характер нашей области означает, что мы решили использовать транзакционные шаблоны для обеспечения семантики «Точно один раз», в некоторых случаях применяя шаблон «потребитель-преобразование-производство».
Мы также изучаем, как это сделать. обрабатывать сценарии сбоев, исследуя, как традиционные механизмы дедлеттеринга могут позволить нам поддерживать выборочное воспроизведение событий. И вот здесь у нас возникла проблема.
В нашем IoC мы регистрируем двух производителей, которые по сути выглядят так (псевдо):

Код: Выделить всё

services.AddSingleton()
services.AddSingleton, MessageProducer>
services.AddSingleton

Как уже отмечалось, вышеприведенное является псевдо-, но конечным желаемым результатом является то, что при успешном некотором внутреннем поведении мы используем MessageProducer и отправляем событие (например, TypeA) в заданную тему ( Тема А). В случае сбоя мы используем другой экземпляр MessageProducer для отправки исходного события, полученного из исходной темы, в {sourcetopic}_dlt.
Каждый экземпляр MessageProducer внутренне вызывает InitTransaction. и иногда это приводит к исключению KafkaException. KakfaConnection, передаваемый каждому экземпляру MessageProducer, является одним и тем же экземпляром KafkcConnection (singleton) и настроен с использованием одного и того же ProducerConfig, включая один и тот же TransactionalId.
Я считаю, что мы вызываем InitTransaction (только один раз), но в каждом экземпляре (2 экземпляра) MessageProducer с одним и тем же TransactionalId, что вызывает КафкаИсключение. Иногда это работает, иногда нет. Но я считаю, что настройка неправильная.
Поэтому у меня три вопроса:

• Какая настройка правильная? ? Прав ли я, полагая, что, несмотря на то, что в моем сервисе есть два логических «производителя», производящих продукцию по двум темам (тема успеха и тема недоставленного письма), существует только один сервис, и поэтому мы должны вызывать InitTransactions только один раз (независимо от сколько у нас логичных производителей). т. е. производитель = развернутый сервис, производящий Kafka.
• Каким должен быть TransactionalId в сценарии с горизонтальным масштабированием. т. е. у меня есть служба, выполняющая роль, но 3 экземпляра этой службы в горизонтально масштабируемом сценарии, причем каждый экземпляр принадлежит одной и той же группе потребителей (следовательно, одной и той же роли). Должен ли каждый экземпляр иметь одинаковый TransactionalId или отдельный TransactionalId?
• Возможно, нам также что-то не хватает при выборочном воспроизведении событий. В подмодели публикации (например, 1 производитель, 2 потребителя) с другими платформами обмена сообщениями проще добиться выборочного воспроизведения только для потребителя, который не смог воспроизвести событие 1-го класса. Я не уверен, насколько легко это сделать с помощью Confluent Kafka. Мы пытаемся придумать несколько подходящих шаблонов, которые не зависят от того, что другие сотрудники обладают высокой квалификацией/знанием Confluent CLI + и должны иметь возможность запускать команды CLI в производственном кластере. Будем благодарны за любые советы.

Документация API отсюда: https://github.com/confluentinc/confluent -kafka-dotnet/blob/master/src/Confluent.Kafka/IProducer.cs#L286
InitTransaction: https://github.com/confluentinc/conflue ... 1-L1048C37

Подробнее здесь: https://stackoverflow.com/questions/792 ... aexception