Что такое latest смещение консьюмера в Kafka?

Что такое latest смещение консьюмера в Kafka

Latest offset — это параметр конфигурации Kafka потребителя, который определяет стратегию чтения сообщений. Когда потребитель подключается к разделу впервые и не имеет сохранённого смещения, параметр auto.offset.reset со значением latest указывает начать чтение с самого последнего сообщения в разделе.

Основные смещения в Kafka

Каждый раздел (partition) имеет несколько важных смещений:

• Beginning offset (начальное) — смещение первого сообщения в разделе
• Latest offset (последнее) — смещение следующего сообщения, которое будет добавлено (текущий размер раздела)
• Committed offset (закоммиченное) — последнее смещение, обработанное потребителем группы
• Current position (текущая позиция) — смещение, с которого потребитель будет читать

Структура смещений

Раздел 0:
┌─────┬─────┬─────┬─────┬─────┬─────┐
│ 0   │ 1   │ 2   │ 3   │ 4   │ 5   │
└─────┴─────┴─────┴─────┴─────┴─────┘
↑                               ↑
Beginning offset            Latest offset

Параметр auto.offset.reset

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Properties;

public class OffsetResetExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
            "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
            "org.apache.kafka.common.serialization.StringDeserializer");
        
        // Если нет сохранённого offset, начать с latest
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));
        
        // При первом подключении будут прочитаны только новые сообщения
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(1000);
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Message: " + record.value() + 
                    ", Offset: " + record.offset());
            }
        }
    }
}

Сценарий: Первый запуск с latest

Час 1: Топик содержит сообщения 0-999
Новый потребитель подключается с auto.offset.reset=latest
  ↓
Потребитель начинает читать с offset 1000 (следующий)
  ↓
Сообщения 0-999 пропускаются полностью
  ↓
Потребитель получает только новые сообщения (1000, 1001, ...)

Различия между latest и earliest

public class OffsetComparisonExample {
    
    // С latest: читаем только новые сообщения
    static void withLatest() {
        Properties props = new Properties();
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        // Сообщения, добавленные ДО подключения, не будут прочитаны
    }
    
    // С earliest: читаем все сообщения с начала
    static void withEarliest() {
        Properties props = new Properties();
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // Сообщения, добавленные раньше, будут прочитаны все
    }
}

Получение текущего offset

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

public class GetOffsetExample {
    public static void main(String[] args) {
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));
        
        // Получаем информацию после первого poll
        consumer.poll(0);
        
        for (TopicPartition partition : consumer.assignment()) {
            // Текущая позиция потребителя
            long currentPosition = consumer.position(partition);
            System.out.println("Current position: " + currentPosition);
            
            // Последнее закоммиченное смещение
            long committedOffset = consumer.committed(partition).offset();
            System.out.println("Committed offset: " + committedOffset);
            
            // Latest offset раздела
            consumer.seekToEnd(partition);
            long latestOffset = consumer.position(partition);
            System.out.println("Latest offset: " + latestOffset);
            
            // Beginning offset раздела
            consumer.seekToBeginning(partition);
            long beginningOffset = consumer.position(partition);
            System.out.println("Beginning offset: " + beginningOffset);
        }
    }
}

Практический пример с отслеживанием offset

public class OffsetTrackingExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "tracking-group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
        
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));
        
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(1000);
            
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(
                    "Topic: " + record.topic() +
                    ", Partition: " + record.partition() +
                    ", Offset: " + record.offset() +
                    ", Value: " + record.value()
                );
            }
            
            // Выводим текущие offset для каждого раздела
            for (TopicPartition partition : consumer.assignment()) {
                long position = consumer.position(partition);
                System.out.println(
                    "Partition " + partition.partition() + 
                    ", Current offset: " + position
                );
            }
        }
    }
}

Когда использовать latest

1. Монитор/alerting система — нужны только новые события
2. Восстановление потребителя — не нужна история
3. Log aggregation — начинаем с текущей позиции
4. Real-time обработка — пропускаем исторические данные

Когда использовать earliest

1. Обработка всей истории — нужны все данные
2. Analytics — аналитика требует полных данных
3. Восстановление состояния — воссоздаём полное состояние
4. Миграция данных — нужно перенести все данные

Понимание latest offset критично для корректной работы с Kafka потребителями и выбора правильной стратегии обработки сообщений.