Apache Kafka
Consumer — группы, оффсеты, ребалансинг
Apache Kafka Consumer — это клиент, ответственный за чтение и обработку сообщений из топиков.
Group coordinator, heartbeat, session.timeout.ms
Consumer groups позволяют нескольким потребителям параллельно обрабатывать топик, распределяя партиции между ними для load balancing. Группа идентифицируется по group.id; каждый потребитель в группе — member с уникальным member.id (генерируется при join).
Group coordinator — это брокер, выбранный для управления группой (hash(group.id) % num_brokers определяет координатора).
Координатор хранит состояние группы: members, assignments партиций, committed offsets (в внутренней теме __consumer_offsets).
В памяти брокера: состояние в GroupMetadataManager, с off-heap структурами для efficiency; offsets реплицированы как обычные записи.
Потребители поддерживают связь через heartbeat'ы: отдельный Heartbeat thread в KafkaConsumer посылает HeartbeatRequest каждые heartbeat.interval.ms (по умолчанию 3 сек) к координатору. Координатор отвечает, подтверждая membership. Если heartbeat не приходит в течение session.timeout.ms (по умолчанию 45 сек), member считается dead, триггеря rebalance.
В памяти потребителя: Coordinator хранит generation (epoch группы), member.id. Нюанс: высокое session.timeout.ms позволяет пережить GC-паузы или network glitches, но замедляет detection failures; низкое — приводит к frequent rebalances. При poll() heartbeat отправляется implicitly, если interval истек.
Offset management: авто-коммит vs ручной (commitSync, commitAsync)
Offsets тракают прогресс чтения: для каждой партиции — committed offset, с которого группа начнет после restart/rebalance.
Авто-коммит: Включается enable.auto.commit=true (default), с auto.commit.interval.ms (5 сек). При poll() потребитель аккумулирует processed offsets в памяти (OffsetAndMetadata), и каждые interval commit'ит их async. В памяти: Map<TopicPartition, OffsetAndMetadata> в KafkaConsumer, flushed periodically.
Trade-offs: удобно, но риск at-most-once (если crash после обработки, но до commit — потеря) или at-least-once (duplicates если commit после обработки, но перед full ack). Не используйте для critical processing.
Ручной коммит: enable.auto.commit=false. commitSync() — synchronous, блокирует до подтверждения от координатора (via OffsetCommitRequest), обеспечивает durability, но увеличивает latency. commitAsync() — asynchronous, с optional callback для error handling; faster, но требует manual retry на failures.
В памяти: offsets буферизуются до commit; при commitSync все in-flight commits ждут. Нюанс: commit только assigned партиций; при rebalance revoked offsets commit'ятся в onPartitionsRevoked. Для EOS используйте с transactions (read-process-write pattern).
#Java #middle #Kafka #Consumer
Consumer — группы, оффсеты, ребалансинг
Apache Kafka Consumer — это клиент, ответственный за чтение и обработку сообщений из топиков.
Group coordinator, heartbeat, session.timeout.ms
Consumer groups позволяют нескольким потребителям параллельно обрабатывать топик, распределяя партиции между ними для load balancing. Группа идентифицируется по group.id; каждый потребитель в группе — member с уникальным member.id (генерируется при join).
Group coordinator — это брокер, выбранный для управления группой (hash(group.id) % num_brokers определяет координатора).
Координатор хранит состояние группы: members, assignments партиций, committed offsets (в внутренней теме __consumer_offsets).
В памяти брокера: состояние в GroupMetadataManager, с off-heap структурами для efficiency; offsets реплицированы как обычные записи.
Потребители поддерживают связь через heartbeat'ы: отдельный Heartbeat thread в KafkaConsumer посылает HeartbeatRequest каждые heartbeat.interval.ms (по умолчанию 3 сек) к координатору. Координатор отвечает, подтверждая membership. Если heartbeat не приходит в течение session.timeout.ms (по умолчанию 45 сек), member считается dead, триггеря rebalance.
В памяти потребителя: Coordinator хранит generation (epoch группы), member.id. Нюанс: высокое session.timeout.ms позволяет пережить GC-паузы или network glitches, но замедляет detection failures; низкое — приводит к frequent rebalances. При poll() heartbeat отправляется implicitly, если interval истек.
Offset management: авто-коммит vs ручной (commitSync, commitAsync)
Offsets тракают прогресс чтения: для каждой партиции — committed offset, с которого группа начнет после restart/rebalance.
Авто-коммит: Включается enable.auto.commit=true (default), с auto.commit.interval.ms (5 сек). При poll() потребитель аккумулирует processed offsets в памяти (OffsetAndMetadata), и каждые interval commit'ит их async. В памяти: Map<TopicPartition, OffsetAndMetadata> в KafkaConsumer, flushed periodically.
Trade-offs: удобно, но риск at-most-once (если crash после обработки, но до commit — потеря) или at-least-once (duplicates если commit после обработки, но перед full ack). Не используйте для critical processing.
Ручной коммит: enable.auto.commit=false. commitSync() — synchronous, блокирует до подтверждения от координатора (via OffsetCommitRequest), обеспечивает durability, но увеличивает latency. commitAsync() — asynchronous, с optional callback для error handling; faster, но требует manual retry на failures.
В памяти: offsets буферизуются до commit; при commitSync все in-flight commits ждут. Нюанс: commit только assigned партиций; при rebalance revoked offsets commit'ятся в onPartitionsRevoked. Для EOS используйте с transactions (read-process-write pattern).
#Java #middle #Kafka #Consumer
👍5
Rebalancing: Range, RoundRobin, Cooperative Sticky
Rebalancing — процесс перераспределения партиций при changes в группе (join/leave, failures). Триггерится coordinator'ом: все members посылают JoinGroupRequest, coordinator выбирает leader (первый joiner), который вычисляет assignments via assignor.
Assignors (partition.assignor.class):
- Range: Дефолт до 2.4. Партиции сортируются, делятся по range (например, для 10 partitions, 3 consumers: 0-3,4-6,7-9). Skew если num_partitions не делится evenly.
- RoundRobin: Распределяет round-robin для fairness, минимизируя skew.
- Cooperative Sticky (default с 2.4): Эволюция Sticky (минимизирует перемещения партиций). Cooperative — incremental: вместо full revoke-assign, использует несколько раундов (Eager vs Cooperative protocol). В первом раунде revoke только conflicting partitions, затем assign. Снижает downtime: consumers продолжают process revoked-later.
В памяти coordinator'а: хранит previous assignments для sticky. Нюанс: custom assignor implement ConsumerPartitionAssignor; для large groups (>100) rebalance может занять секунды из-за sync.
Static membership
Static membership избегает rebalance при restarts: group.instance.id (уникальный static ID per instance). При join с тем же ID, coordinator распознает как restart, не триггеря rebalance (если assignments unchanged).
В памяти: coordinator тракает instances в GroupMetadata. Нюанс: полезно для stateful consumers (с local state); но если instance меняет host, rebalance все равно. Комбинируйте с cooperative для minimal disruption.
Pause/Resume и backpressure
Для контроля flow: consumer.pause(Collection<TopicPartition>) останавливает fetch для партиций, но poll() продолжает heartbeat. resume() возобновляет. Используйте для backpressure: если downstream slow, pause до обработки backlog.
В памяти: paused partitions в Set<TopicPartition> в Fetcher; fetch requests skip them. Нюанс: paused не влияет на rebalance; при длительном pause lag растет, рискуя eviction из группы если max.poll.interval.ms истекает.
Backpressure: мониторьте lag, dynamically pause если queue full. В Streams это built-in via task pausing.
Метрики: lag, rebalance-latency
Consumer экспортирует метрики via KafkaConsumer.metrics():
- consumer-lag: Records-lag-max — максимальный lag (LEO - committed offset) по партициям. Вычисляется via FetchRequest с metadata. Высокий lag указывает на slow processing; мониторьте per-partition.
- rebalance-latency-avg: Среднее время rebalance (от trigger до completion). Включает join, sync, assign. Высокое — из-за large groups или slow assignors.
Другие: poll-time, commit-latency. Нюанс: используйте ConsumerMetrics для JMX; в production alert на lag > threshold или frequent rebalances.
#Java #middle #Kafka #Consumer
Rebalancing — процесс перераспределения партиций при changes в группе (join/leave, failures). Триггерится coordinator'ом: все members посылают JoinGroupRequest, coordinator выбирает leader (первый joiner), который вычисляет assignments via assignor.
Assignors (partition.assignor.class):
- Range: Дефолт до 2.4. Партиции сортируются, делятся по range (например, для 10 partitions, 3 consumers: 0-3,4-6,7-9). Skew если num_partitions не делится evenly.
- RoundRobin: Распределяет round-robin для fairness, минимизируя skew.
- Cooperative Sticky (default с 2.4): Эволюция Sticky (минимизирует перемещения партиций). Cooperative — incremental: вместо full revoke-assign, использует несколько раундов (Eager vs Cooperative protocol). В первом раунде revoke только conflicting partitions, затем assign. Снижает downtime: consumers продолжают process revoked-later.
В памяти coordinator'а: хранит previous assignments для sticky. Нюанс: custom assignor implement ConsumerPartitionAssignor; для large groups (>100) rebalance может занять секунды из-за sync.
Static membership
Static membership избегает rebalance при restarts: group.instance.id (уникальный static ID per instance). При join с тем же ID, coordinator распознает как restart, не триггеря rebalance (если assignments unchanged).
В памяти: coordinator тракает instances в GroupMetadata. Нюанс: полезно для stateful consumers (с local state); но если instance меняет host, rebalance все равно. Комбинируйте с cooperative для minimal disruption.
Pause/Resume и backpressure
Для контроля flow: consumer.pause(Collection<TopicPartition>) останавливает fetch для партиций, но poll() продолжает heartbeat. resume() возобновляет. Используйте для backpressure: если downstream slow, pause до обработки backlog.
В памяти: paused partitions в Set<TopicPartition> в Fetcher; fetch requests skip them. Нюанс: paused не влияет на rebalance; при длительном pause lag растет, рискуя eviction из группы если max.poll.interval.ms истекает.
Backpressure: мониторьте lag, dynamically pause если queue full. В Streams это built-in via task pausing.
Метрики: lag, rebalance-latency
Consumer экспортирует метрики via KafkaConsumer.metrics():
- consumer-lag: Records-lag-max — максимальный lag (LEO - committed offset) по партициям. Вычисляется via FetchRequest с metadata. Высокий lag указывает на slow processing; мониторьте per-partition.
- rebalance-latency-avg: Среднее время rebalance (от trigger до completion). Включает join, sync, assign. Высокое — из-за large groups или slow assignors.
Другие: poll-time, commit-latency. Нюанс: используйте ConsumerMetrics для JMX; в production alert на lag > threshold или frequent rebalances.
#Java #middle #Kafka #Consumer
👍5
Пример кода
Пример на Java с subscribe и RebalanceListener для handling revoke/assign:
В production: implement commitOffsets() с Map<TopicPartition, OffsetAndMetadata>; handle exceptions в listener.
Нюансы
- Избегание frequent rebalancing: Frequent rebalances (от scaling, GC, network) приводят к downtime (poll blocks во время). Static membership стабилизирует группу при restarts. CooperativeStickyAssignor минимизирует перемещения (до 80% меньше чем Range). Увеличьте session.timeout.ms до 300 сек для tolerance; heartbeat.interval.ms= session/3. Мониторьте rebalance-rate; если high — investigate app stability.
- max.poll.interval.ms и долгие операции: По умолчанию 5 мин — максимальное время между poll(). Если processing в poll() превышает (например, heavy computation), coordinator считает dead, триггеря rebalance. Решение: разбейте работу на chunks, poll() frequently; используйте pause() для long ops, но resume timely. Для очень долгих — offload в separate thread, но sync с poll(). Нюанс: в Streams это processing.guarantee=at_least_once handles.
- Обработка с сохранением ordering: Consumer гарантирует order только внутри партиции, но rebalance может нарушить если state не сохранен. В onPartitionsRevoked commit offsets и persist state (например, в external store). В onPartitionsAssigned seek(committed offset) и restore state. Для strict ordering: single-threaded per partition, или assign manually (без subscribe, используйте assign()). Нюанс: в groups с multiple consumers ordering cross-partition не гарантировано; для global order — single partition или external sorting.
#Java #middle #Kafka #Consumer
Пример на Java с subscribe и RebalanceListener для handling revoke/assign:
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.util.*;
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false"); // Manual commit
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("topic"), new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
// Commit offsets before losing partitions
commitOffsets(); // Custom method to commit processed offsets
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
// Restore state or seek to offsets
/* warm-up: e.g., load local state for partitions */
}
});
try {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
// Process records
consumer.commitAsync(); // Or commitSync for sync
}
} finally {
consumer.close();
}
В production: implement commitOffsets() с Map<TopicPartition, OffsetAndMetadata>; handle exceptions в listener.
Нюансы
- Избегание frequent rebalancing: Frequent rebalances (от scaling, GC, network) приводят к downtime (poll blocks во время). Static membership стабилизирует группу при restarts. CooperativeStickyAssignor минимизирует перемещения (до 80% меньше чем Range). Увеличьте session.timeout.ms до 300 сек для tolerance; heartbeat.interval.ms= session/3. Мониторьте rebalance-rate; если high — investigate app stability.
- max.poll.interval.ms и долгие операции: По умолчанию 5 мин — максимальное время между poll(). Если processing в poll() превышает (например, heavy computation), coordinator считает dead, триггеря rebalance. Решение: разбейте работу на chunks, poll() frequently; используйте pause() для long ops, но resume timely. Для очень долгих — offload в separate thread, но sync с poll(). Нюанс: в Streams это processing.guarantee=at_least_once handles.
- Обработка с сохранением ordering: Consumer гарантирует order только внутри партиции, но rebalance может нарушить если state не сохранен. В onPartitionsRevoked commit offsets и persist state (например, в external store). В onPartitionsAssigned seek(committed offset) и restore state. Для strict ordering: single-threaded per partition, или assign manually (без subscribe, используйте assign()). Нюанс: в groups с multiple consumers ordering cross-partition не гарантировано; для global order — single partition или external sorting.
#Java #middle #Kafka #Consumer
👍3🤯2