Уточнение кафки

Question

Я использую Spring-Kafka 1.3.5, который использует клиент 0.11.0.2. У меня есть 2 раздела. И мой max.poll.records = 1.

У меня такой вопрос: предположим, у меня есть 2 узла / работника (для которых параллелизм установлен на один, поэтому один потребитель в каждом узле), которые являются медленными работниками (может потребоваться 5 минут для обработки записей), оба в настоящее время отключены. Когда узел-1 (потребитель-1) присоединяется к группе потребителей, ему назначаются 2 раздела (p-1, p-2). Теперь потребитель-1 потребляет сообщение от p-1. И сразу же узел-2 (потребитель-2) присоединяется к той же группе потребителей.

1. Есть ли вероятность, что p-1 будет назначен для потребителя-2, или kafka знает, что потребитель-1 в настоящее время обрабатывает сообщение, поэтому он назначает раздел-2 для потребителя-2? Если это может произойти, может ли узел-2 извлечь то же сообщение (поскольку оно еще не подтверждено) из p-1, и когда потребитель-1 завершит обработку и когда попытается зафиксировать свое смещение, будет ли исключение CommmitFailedException или какое-либо исключение бросили?

2. Может ли потребитель вызвать syncCommits() для раздела, которым он больше не владеет? И будет ли выброшено исключение?

Answer 1

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

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

Для назначения разделов у Кафки есть три стратегии назначения разделов - Range и RoundRobin и Sticky Assignor. Вы можете прочитать больше об этом здесь.

Присвоение диапазона - https://kafka.apache.org/10/javadoc/index.html?org/apache/kafka/clients/consumer/RangeAssignor.html

Круглый Robin Assignor - https://kafka.apache.org/10/javadoc/org/apache/kafka/clients/consumer/RoundRobinAssignor.html

Sticky Assignor - https://kafka.apache.org/10/javadoc/org/apache/kafka/clients/consumer/StickyAssignor.html

Что касается ваших двух вышеупомянутых вопросов, то для потребителя 2 вполне возможно получить назначенный раздел 1 и, следовательно, получить сообщение, которое уже обработано. И если вы выполняете фиксацию смещения вручную, да, вы должны получить исключение CommitFailed, когда потребитель 1 пытается зафиксировать свое смещение, независимо от того, принадлежит ему P1 или нет.