Как я потерял сообщения из Kafka
Однажды на работе столкнулся с неприятным багом - часть сообщений просто не доезжала до сервиса.
Написал consumer с commit после обработки сообщений, казалось, всё ок. Но часть сообщений до сервиса не доезжала. Источник работает по at-least-once, в Kafka сообщения точно попадали.
Начал разбираться, как оказалось, по дефолту в Kafka включён автокоммит:
enable.auto.commit=true # Включить автокоммит
auto.commit.interval.ms=5000 # Автокоммит раз в 5 секунд
Какая здесь проблема?
1. Consumer получил пачку сообщений
2. Kafka автоматически закоммитила offset
3. Посередине обработки случился OOM или Deploy сервиса
4. После рестарта consumer начал читать с уже закоммиченного offset
5. Часть сообщений исчезла навсегда
По итогу отключил автокоммит через enable.auto.commit=false и сообщения перестали теряться.
Однажды на работе столкнулся с неприятным багом - часть сообщений просто не доезжала до сервиса.
Написал consumer с commit после обработки сообщений, казалось, всё ок. Но часть сообщений до сервиса не доезжала. Источник работает по at-least-once, в Kafka сообщения точно попадали.
Начал разбираться, как оказалось, по дефолту в Kafka включён автокоммит:
enable.auto.commit=true # Включить автокоммит
auto.commit.interval.ms=5000 # Автокоммит раз в 5 секунд
Какая здесь проблема?
1. Consumer получил пачку сообщений
2. Kafka автоматически закоммитила offset
3. Посередине обработки случился OOM или Deploy сервиса
4. После рестарта consumer начал читать с уже закоммиченного offset
5. Часть сообщений исчезла навсегда
По итогу отключил автокоммит через enable.auto.commit=false и сообщения перестали теряться.