Какие типичные проблемы могут возникнуть при чтении сообщений из топиков Apache Kafka?

При чтении данных из Kafka потребители (consumers) могут столкнуться с рядом типичных проблем.

Основные проблемы и их причины:

1. Дублирование сообщений (Duplicate Processing):

   • Причина: Потребитель обработал сообщение, но не зафиксировал смещение (offset) перед сбоем. После перезапуска чтение начнется со старого смещения.
   • Решение: Использовать идиоматическую обработку с commit'ом offset'а после успешной обработки. Рассмотреть семантику "точно один раз" (exactly-once) при необходимости.

2. Потеря сообщений (Message Loss):

   • Причина: Автоматический commit offset'а (enable.auto.commit=true) до фактической обработки сообщения. Если потребитель упал после commit'а, но до обработки, сообщение будет потеряно.
   • Решение: Отключить авто-коммит (enable.auto.commit=false) и вручную управлять offset'ами после успешной обработки.

3. Застревание потребителя (Consumer Lag / Stuck Consumer):

   • Причина: Долгая обработка одного сообщения блокирует весь поток. Неправильная настройка max.poll.interval.ms.
   • Решение: Увеличить max.poll.interval.ms, оптимизировать логику обработки, использовать многопоточную обработку внутри потребителя.

4. Ребалансировка потребителей (Rebalancing):

   • Причина: Добавление или удаление потребителей в группе приводит к перераспределению разделов (partitions). Во время ребалансировки чтение приостанавливается.
   • Решение: Минимизировать время ребалансировки, корректно обрабатывать callback'и ConsumerRebalanceListener для ручного управления offset'ами.

5. Некорректная десериализация (Deserialization Errors):

   • Причина: Формат данных в топике не соответствует ожидаемому десериализатором потребителя.
   • Решение: Настроить обработчики ошибок десериализации (default.deserialization.exception.handler), использовать схемы (Avro, Protobuf) с Registry (Confluent Schema Registry).

Пример настройки для избежания потерь и дубликатов (Java):

Properties props = new Properties();
props.put("enable.auto.commit", "false"); // Важно!
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("group.id", "my-consumer-group");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("my-topic"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            // 1. Обработать сообщение
            processMessage(record.value());
            // 2. Вручную закоммитить offset для этой партиции
            Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
            offsets.put(new TopicPartition(record.topic(), record.partition()),
                        new OffsetAndMetadata(record.offset() + 1)); // +1 для следующего
            consumer.commitSync(offsets);
        }
    }
} finally {
    consumer.close();
}