Как настраивается удаление сообщений в Kafka?

В Apache Kafka сообщения хранятся в логах (топиках), которые сегментируются на части. Удаление сообщений происходит не поштучно, а целыми сегментами лога, когда срабатывают заданные политики хранения. Это фундаментальный аспект управления дисковым пространством и соответствия требованиям хранения данных.

Основные механизмы очистки (Cleanup Policies)

Kafka поддерживает две основные политики очистки, задаваемые параметром cleanup.policy топика:

• delete (удаление по времени/размеру): Стандартная политика для большинства топиков. Сообщения удаляются при превышении лимитов времени или размера.
• compact (компактизация): Сохраняет только последнее значение для каждого ключа. Используется для топиков с состоянием (например, changelog).

Ключевые параметры конфигурации

Для политики delete настройки задаются на уровне топика или глобально для брокера:

• retention.ms: Время хранения сообщений в миллисекундах (по умолчанию 7 дней).
• retention.bytes: Максимальный размер топика в байтах до начала удаления старых сегментов.
• log.retention.check.interval.ms: Как часто брокер проверяет сегменты на соответствие политикам (по умолчанию 5 минут).

Пример настройки при создании топика

# Создание топика с политикой удаления через утилиту kafka-topics
kafka-topics.sh --create \
  --topic user-events \
  --partitions 3 \
  --replication-factor 2 \
  --config retention.ms=86400000 \
  --config cleanup.policy=delete \
  --bootstrap-server localhost:9092

# Топик будет хранить сообщения ровно 1 день (86 400 000 мс)

Как происходит процесс удаления

Брокер Kafka запускает фоновую задачу (log cleaner), которая периодически сканирует сегменты логов. Для каждого сегмента проверяется:

1. Время последнего сообщения в сегменте: если оно старше, чем retention.ms.
2. Общий размер лога топика: если он превышает retention.bytes (удаляются самые старые сегменты).

Удаление происходит на уровне файлов сегментов (.log и .index файлы). Активный сегмент (куда идёт текущая запись) никогда не удаляется, даже если его данные устарели.

Особые случаи и тонкости

• Компактизация (compaction): Для топиков с ключами можно включить cleanup.policy=compact. Kafka будет сохранять только последнее сообщение для каждого ключа, удаляя старые дубликаты. Это полезно для топиков-логов изменений (changelog).
• Ручное удаление: Можно удалить сообщения до определённого смещения (offset) с помощью утилиты kafka-delete-records, но это операция для администрирования, а не для регулярного использования.
• Влияние на производительность: Слишком частая проверка (log.retention.check.interval.ms) может нагружать брокер. Слишком редкая — приведёт к задержке в удалении.

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