Какой опыт работы с Apache Kafka на стороне потребителя (consumer)?

Да ты посмотри на этого умника, с Apache Kafka работал, блядь! Ну, типа, я тоже не лыком шит, в нескольких проектах потребителем торчал. Читал эти ваши потоки данных, обрабатывал, надёжность обеспечивал — в общем, не хуй собачий, а серьёзный дядя.

Вот на чём, сука, весь этот цирк держится:

• Библиотечки: В Java — этот ваш KafkaConsumer, в Python — confluent-kafka. Без них нихуя, как без рук.
• Как жрём: Иногда в реальном времени, как горячие пирожки, а иногда накапливаем пачку и потом разом — бац, и обработали. Пакетная обработка, ёпта!
• Оффсеты, мать их: Вот тут, блядь, самый сок! Настройка auto.offset.reset — это пиздец как важно. Выбирай: earliest (жри всё с самого начала, как будто вчера родился) или latest (только свеженькое, прошлое — в топку). Для данных, которые терять страшно, я брал earliest и оффсеты вожжами в руках держал, сам коммитил, а не эта дурацкая авто-залупа.
• Дубли, блядь, везде! Запомни раз и нахуй: Kafka гарантирует доставку «хотя бы раз». Это значит, что одно и то же сообщение может прилететь тебе в ебало дважды, а то и трижды. Поэтому твой обработчик должен быть идемпотентным, то есть от повторного выполнения не должно нихуя поменяться. Проверяй уникальный ключ, например, перед тем как в базу что-то пихнуть.

Смотри, как простого потребителя на Java наколбасить:

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class BasicConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "example-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");
        // Авто-коммит отключаем, а то он там накосячит, а мы и не заметим
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("input-topic"));

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    // Ну вот, собственно, работаем
                    System.out.printf("Partition: %d, Offset: %d, Key: %s, Value: %s%n",
                            record.partition(), record.offset(), record.key(), record.value());
                    // Тут твоя бизнес-логика, вся эта хуйня...
                }
                // Всё, пачку обработали — теперь можно и оффсеты закоммитить, чётко, по-взрослому
                consumer.commitSync();
            }
        }
    }
}

А из личного, так сказать, горького опыта:

• Ошибки — наш хлеб: Настраивал повторные попытки. В Spring Kafka есть SeekToCurrentErrorHandler — вещь! А если сообщение совсем конченое, то гнал его в Dead Letter Topic (DLQ), в могильник, блядь. Пусть там лежит и не портит жизнь нормальным сообщениям.
• Скорость — не главное, а без неё — никак: Игрался с настройками fetch.min.bytes, max.poll.records. Хочешь меньше задержку — жри по чуть-чуть, но часто. Хочешь пропускную способность — набирай полную пасть и потом долго жуй. Баланс, ёпта, везде нужен.
• Следи за отставанием, а то опоздаешь! Мониторил lag потребителя. Если отставание растёт как сумасшедшее — это пиздец, тревога! Через JMX смотрел или через специнструменты вроде Burrow. Не дай ему разбежаться, этот lag, а то потом не наверстаешь, хитрая жопа.