Что является первичным ключом оффсета в Kafka?

Что является первичным ключом оффсета в Kafka?

В Apache Kafka первичным ключом для хранения оффсета является комбинация трёх элементов:

Структура первичного ключа оффсета

Первичный ключ = (Topic, Partition, Consumer Group)

• Topic — название топика
• Partition — номер партиции (0, 1, 2, ...)
• Consumer Group — название группы потребителей

Эта комбинация уникально идентифицирует позицию в потоке сообщений для конкретной группы потребителей.

Внутренняя структура хранения оффсета

Когда Kafka хранит оффсеты, она использует специальный внутренний топик __consumer_offsets с ключом, структурированным следующим образом:

Key: consumer_group_id:topic:partition
Value: offset_value

Практический пример

Представим, что у нас есть:

• Topic: "orders"
• Partition: 0, 1, 2
• Consumer Group: "payment-processors"

Для этой конфигурации Kafka будет хранить три оффсета:

ключ: payment-processors:orders:0 → значение: 1000
ключ: payment-processors:orders:1 → значение: 850
ключ: payment-processors:orders:2 → значение: 920

Это означает, что группа потребителей "payment-processors" прочитала:

• 1000 сообщений из партиции 0
• 850 сообщений из партиции 1
• 920 сообщений из партиции 2

Где хранятся оффсеты

Версия Kafka < 0.10: Оффсеты хранились в Apache ZooKeeper под путём /consumers/{group}/offsets/{topic}/{partition}

Версия Kafka >= 0.10: Оффсеты хранятся в внутреннем топике __consumer_offsets. Это более надёжно и масштабируемо.

Как просмотреть оффсеты

# Посмотреть текущие оффсеты группы
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group payment-processors \
  --describe

# Вывод:
# TOPIC         PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# orders        0          1000            1050            50
# orders        1          850             900             50
# orders        2          920             980             60

Commits и оффсеты

Потребитель может использовать разные стратегии коммита оффсета:

Автоматический коммит:

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    orders,
    bootstrap_servers=[localhost:9092],
    group_id=payment-processors,
    auto_offset_reset=latest,
    enable_auto_commit=True,  # Автокоммит каждые 5 сек
    auto_commit_interval_ms=5000
)

for message in consumer:
    process_order(message.value)
    # Оффсет автоматически закоммитится

Ручной коммит:

consumer = KafkaConsumer(
    orders,
    bootstrap_servers=[localhost:9092],
    group_id=payment-processors,
    enable_auto_commit=False  # Отключаем автокоммит
)

for message in consumer:
    try:
        process_order(message.value)
        consumer.commit()  # Коммитим только после успешной обработки
    except Exception as e:
        log_error(e)
        # Если ошибка, оффсет не коммитится и сообщение будет обработано снова

Важные моменты оффсетов

Изоляция по группам — каждая группа потребителей имеет свой оффсет, поэтому группы не влияют друг на друга.

Переиспользование сообщений — если потребитель упал и не закоммитил оффсет, при перезагрузке он начнёт с последнего закоммиченного оффсета и переобработает сообщения.

Consumer Lag — разница между LOG-END-OFFSET (последнее сообщение) и CURRENT-OFFSET (где мы сейчас). Показывает, на сколько потребитель отстаёт.

LAG = LOG-END-OFFSET - CURRENT-OFFSET

Сброс оффсетов

Если нужно переобработать сообщения, можно сбросить оффсеты:

# Сброс на начало топика
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group payment-processors \
  --topic orders \
  --reset-offsets --to-earliest --execute

# Сброс на конец топика
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group payment-processors \
  --topic orders \
  --reset-offsets --to-latest --execute

Оффсеты в Kafka — критический элемент гарантирования обработки сообщений и восстановления после сбоев. Правильное управление оффсетами обеспечивает надёжную и предсказуемую работу потребителей.