Как настроить Kafka, чтобы для пользователя сохранялось только последнее событие?

Как сохранить в Kafka только последнее событие пользователя?

• использовать ключи сообщений (key) — в качестве ключа передавать идентификатор пользователя
• включить для топика compaction — Kafka сохранит только последнее сообщение с уникальным ключом
• при compaction старые дубликаты одного ключа регулярно удаляются, поэтому остаётся его последнее значение
• указать параметр топика: cleanup.policy=compact
• подход подходит для случаев, когда требуется актуальное состояние, а не полный журнал событий
• важное ограничение: чтобы удалить записи, необходимо отправлять сообщения с null-значением — так называемые tombstone
• подход применяют для кеширования состояния или репликации исходного состояния

Итак, решение заключается в использовании топика с compaction по ключу пользователя: в нём сохраняется только последнее актуальное событие.

Подробный ответ

Основной ответ

Чтобы Apache Kafka хранила для каждого пользователя только последнее событие, нужно задействовать ключ сообщения (message key) и настроить топик как compacted topic. При включённой компaction Kafka удаляет прежние записи с одинаковым ключом и сохраняет последнее значение, благодаря чему для каждого ключа остаётся наиболее новое событие.

Ключевые моменты

• Применение ключа сообщения: каждому пользовательскому событию назначается уникальный ключ, например userId. Все последующие изменения данных этого пользователя публикуются с тем же ключом.
• Настройка compacted топика: благодаря параметру cleanup.policy=compact Kafka хранит для каждого ключа только последнюю версию события, удаляя предыдущие версии в фоновом режиме.
• Компромисс между latency и storage: компактация выполняется асинхронно, поэтому старые версии ещё некоторое время могут находиться в топике. Однако со временем они удаляются, и топик сокращается до актуальных данных.
• Для более эффективного использования диска дополнительно настраивают параметры компрессии и retention.

Практический контекст

Этот вариант часто используют для хранения состояния пользователя или его профиля, когда важны текущие данные, а не вся история событий. Например, микросервисы публикуют изменения пользовательских данных с userId в роли ключа, а для топика задают политику compact. В результате можно реализовать эффективный event sourcing с минимальным объёмом хранилища, сохранив актуальные данные.