Как узнать, что сообщение прочитано Kafka

1️⃣ Как кратко ответить

В Kafka сообщение считается прочитанным, когда потребитель успешно обработал его и подтвердил это с помощью механизма подтверждения (offset commit). Потребитель обновляет смещение (offset) в Kafka, указывая, что он обработал сообщение, и это смещение сохраняется в Kafka для отслеживания прогресса.

2️⃣ Подробное объяснение темы

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

Чтобы понять, как Kafka определяет, что сообщение прочитано, нужно разобраться в механизме смещений (offsets) и подтверждений (commits).

Смещения (Offsets)

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

Подтверждения (Commits)

Когда потребитель читает сообщение из партиции, он может подтвердить, что сообщение было обработано. Это делается путем сохранения текущего смещения в специальной внутренней теме Kafka, называемой __consumer_offsets. Это действие называется "commit" смещения.

Как это работает

1. Чтение сообщения: Потребитель читает сообщение из партиции. Он использует текущее смещение, чтобы знать, с какого места продолжить чтение.

2. Обработка сообщения: Потребитель выполняет необходимые действия с сообщением, например, обрабатывает данные или сохраняет их в базе данных.

3. Подтверждение смещения: После успешной обработки сообщения потребитель обновляет смещение, указывая, что это сообщение было прочитано и обработано. Это смещение сохраняется в Kafka.

Пример кода

Рассмотрим пример на Go с использованием библиотеки sarama, популярной для работы с Kafka:

package main
​
import (
    "log"
    "github.com/Shopify/sarama"
)
​
func main() {
    // Создаем конфигурацию потребителя
    config := sarama.NewConfig()
    config.Consumer.Offsets.Initial = sarama.OffsetOldest // Начинаем с самого старого сообщения
​
    // Подключаемся к Kafka
    consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, config)
    if err != nil {
        log.Fatalln("Failed to start consumer:", err)
    }
    defer consumer.Close()
​
    // Подписываемся на партицию топика
    partitionConsumer, err := consumer.ConsumePartition("example_topic", 0, sarama.OffsetNewest)
    if err != nil {
        log.Fatalln("Failed to start partition consumer:", err)
    }
    defer partitionConsumer.Close()
​
    // Читаем сообщения
    for message := range partitionConsumer.Messages() {
        log.Printf("Message claimed: value = %s, offset = %d", string(message.Value), message.Offset)
​
        // Здесь происходит обработка сообщения
​
        // Подтверждаем смещение
        // В sarama это делается автоматически, но можно управлять вручную
    }
}

• Создание конфигурации потребителя: Настраиваем потребителя, указывая, с какого смещения начинать чтение (OffsetOldest для самого старого сообщения).
• Подключение к Kafka: Создаем нового потребителя, подключаясь к брокеру Kafka.
• Подписка на партицию: Подписываемся на конкретную партицию топика.
• Чтение сообщений: В цикле читаем сообщения из партиции. Каждое сообщение имеет смещение, которое мы можем использовать для подтверждения.
• Обработка и подтверждение: Обрабатываем сообщение и подтверждаем смещение. В sarama подтверждение смещения происходит автоматически, но можно управлять этим процессом вручную, если требуется.

Зачем это нужно

Подтверждение смещений позволяет гарантировать, что сообщения не будут потеряны и не будут обработаны повторно. Это важно для обеспечения надежности и согласованности данных в системах, использующих Kafka для передачи сообщений.