Consumer API

KafkaConsumer — это основной интерфейс для чтения сообщений из Kafka. В отличие от push-модели (как в RabbitMQ, где брокер доставляет сообщения), Kafka использует pull-модель: consumer сам запрашивает данные от брокера, контролируя темп чтения. Это решение оказалось ключевым для масштабируемости — брокер не отслеживает скорость каждого потребителя.

KafkaConsumer: конфигурация и создание

Конструктор KafkaConsumer принимает несколько ключевых параметров, определяющих поведение при чтении.

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    bootstrap_servers=['localhost:9092'],
    group_id='analytics',
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    # group.protocol=consumer для KIP-848 (Kafka 4.0+)
)

subscribe vs assign: два режима подписки

Consumer может получать сообщения одним из двух способов. Выбор определяет, кто управляет назначением партиций.

from kafka import KafkaConsumer, TopicPartition

# Режим 1: subscribe — group-managed (рекомендуется)
consumer.subscribe(topics=['events', 'orders'])

# Режим 2: assign — ручное назначение конкретных партиций
consumer.assign(partitions=[
    TopicPartition('events', 0),
    TopicPartition('events', 1),
])

Когда использовать assign: при реализации специализированного consumer (Kafka Streams, consumer, который должен всегда читать конкретную партицию) или в тестовом коде. В большинстве production-сценариев используется subscribe.

poll() loop: сердце consumer

Метод poll() — это единственный способ получить сообщения от Kafka. Он блокирует вызывающий поток на время до timeout_ms миллисекунд, ожидая данных.

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    bootstrap_servers=['localhost:9092'],
    group_id='analytics',
    auto_offset_reset='earliest',
)
consumer.subscribe(['events'])

try:
    while True:
        # poll() возвращает dict {TopicPartition: [ConsumerRecord]}
        records = consumer.poll(timeout_ms=1000, max_records=100)

        for tp, messages in records.items():
            for msg in messages:
                # Обработка каждого сообщения
                print(f'Topic: {msg.topic}, Partition: {msg.partition}, '
                      f'Offset: {msg.offset}, Value: {msg.value}')

        # Offset коммитится автоматически (enable_auto_commit=True)
finally:
    consumer.close()

ConsumerRecord: поля сообщения

Каждое сообщение возвращается как объект ConsumerRecord со следующими полями:

for tp, messages in records.items():
    for record in messages:
        print(f"Топик: {record.topic}")
        print(f"Партиция: {record.partition}, Offset: {record.offset}")
        print(f"Ключ: {record.key}")
        print(f"Значение: {record.value}")
        print(f"Timestamp: {record.timestamp}")

Жизненный цикл consumer

Правильная инициализация и завершение работы consumer — критически важны для корректного управления группой:

1. Создать — KafkaConsumer(bootstrap_servers=..., group_id=...)
2. Подписаться — subscribe(topics) или assign(partitions)
3. Poll в цикле — основная работа, регулярно вызывать poll()
4. Обработать — бизнес-логика для каждого ConsumerRecord
5. Commit — зафиксировать прогресс (авто или ручной)
6. Close — consumer.close() для корректного выхода из consumer group и освобождения ресурсов

Итог

Consumer API строится вокруг простой модели: создать → подписаться → poll в цикле → обработать → закрыть. poll() — единственный способ получить данные. subscribe() делегирует назначение партиций группе, assign() даёт полный контроль. ConsumerRecord содержит всю метаинформацию, необходимую для обработки и управления offset.