Блог

Я разрабатывал потребительское приложение kafka (C# в kubernetes) и некоторое время запускал его как один узел, используя одну тему.

Сегодня я заметил, что тема, из которой я потреблял, была довольно полной — я выполнял непрерывную обработку и имел смещения около ~38 тыс. (в общем, не зависел от раздела), но записи, которые мой продюсер включал в тему (также игнорируя различия в разделах), были около смещений ~58 тыс.

Я решил расширить другой модуль для потребителей — тот же код и конфигурация (идентификатор группы и т. Д.)

Когда он вышел в Сеть, он зарегистрировал, что обрабатывает сообщения в диапазоне смещения ~58 тыс. Я подумал, что это, возможно, просто другой раздел, но я могу видеть один и тот же раздел в обоих журналах (с разными смещениями).

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

Другими словами, почему бы моему второму (или n-му) потребителю не выйти в Сеть и не обработать сообщения в том же диапазоне смещения, что и моему первому потребителю, который работает уже несколько дней?

Я посмотрел некоторые настройки IConsumer, такие как:

• https://docs.confluent.io/platform/current/clients/confluent-kafka-dotnet/_site/api/Confluent.Kafka.ConsumerConfig.html#Confluent_Kafka_ConsumerConfig_QueuedMinMessages

который, по-видимому, указывает минимальное количество сообщений для хранения в «локальной очереди потребителей» (по умолчанию: 100 000), но я не знаю, действительно ли это указывает на то, что ConsumerA претендует на 100 тыс. сообщений, а ConsumerB, естественно, начинает 100 тыс. в будущем

Другие примечания:

• То, что у меня ограниченный доступ к инструментам администрирования (Центр управления), показывает, что мой идентификатор группы потребителей отстает примерно на 900 тысяч сообщений.
• Центр управления говорит, что в моей теме 60 разделов
• Автоматическая фиксация не отключена (по умолчанию: true)
• Независимо от настройки автоматической фиксации, я все еще выполняю _consumer.Фиксация(msg) в блоке finally{} после обработки каждого отдельного сообщения
• Я не хочу убивать своего давнего потребителя (который все еще обрабатывает как чемпион) в случае, если возникнет проблема с сохранением смещения, и я «пропущу» все сообщения в дельте между этими двумя