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

Question

У меня есть потребитель Kafka, который я создаю по расписанию.Он пытается использовать все новые сообщения, которые были добавлены с момента последней фиксации.

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

У меня проблемы с поиском решениячерез документацию Кафки.

Я вижу ряд связанных с тайм-аутом свойств, доступных в классах Confluent.Kafka.ConsumerConfig и ClientConfig, включая FetchWaitMaxMs, но не могу расшифровать, какой из них использовать.Я использую .NET клиент.

Любой совет будет оценен.

Answer 1

Я нашел решение.Версия 1.0.0-бета2 библиотеки Confluent .NET Kafka предоставляет метод с именем .Consume(TimeSpan timeSpan).Это возвратит нуль, если нет новых сообщений для потребления или если мы находимся в разделе EOF.Ранее я использовал перегрузку .Consume(CancellationToken cancellationToken), которая блокировала и не позволяла отключить потребителя.Подробнее здесь: https://github.com/confluentinc/confluent-kafka-dotnet/issues/614#issuecomment-433848857

Другим вариантом было обновление до версии 1.0.0-бета3, которая предоставляет логический флаг для объекта ConsumeResult, называемого IsPartitionEOF.Это то, что я искал изначально - способ узнать, когда я достиг конца раздела.

Answer 2

Я никогда не пользовался клиентом .NET, но при условии, что он не может сильно отличаться от клиента Java, метод poll() должен принимать значение тайм-аута в миллисекундах, поэтому установка в 5000 должна работать в большинстве случаев,Не нужно возиться с классами конфигурации.

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