Модуль 13 ⏱ ~40 мин чтения 📐 5 Mermaid диаграмм

8. Retention и compaction

Управление жизненным циклом и уплотнение лога

«Структуры данных и алгоритмы — это инструменты управления сложностью, но именно управление временем жизни данных сохраняет системам жизнь».
— Брайан Керниган

Два взгляда на жизненный цикл данных

Дисковое пространство любого промышленного сервера конечно. В высоконагруженных системах, генерирующих сотни гигабайт или терабайты событий в сутки, бесконтрольно растущий лог — это верный рецепт аварийной остановки кластера с ошибкой No space left on device (ENOSPC).

При этом характер хранимых данных в бэкенд-архитектурах принципиально различен:

Непрерывный поток неизменяемых событий (Event Stream): Клики пользователей, телеметрия IoT-датчиков, финансовые транзакции, трейсы и логи приложений. В таких сценариях важна возможность воспроизвести поток за последние $N$ часов или дней (например, для расследования инцидентов или повторного обучения моделей машинного обучения), а всё, что старше этого окна, можно безболезненно уничтожить.
Поток изменений состояния сущностей (State Updates / CDC): Балансы пользователей, каталог цен товаров, текущий статус доставки заказа, служебный реестр смещений консьюмеров. Здесь бизнес-логике не интересна вся цепочка из 50 промежуточных изменений цены за полгода — приложению критически важно иметь самое последнее актуальное состояние для каждого первичного ключа!

Apache Kafka предлагает два взаимодополняющих режима очистки лога — delete (Retention) и compact (Log Compaction), — и позволяет гибко настраивать их на уровне каждого отдельного топика через параметр cleanup.policy.

flowchart TD
    classDef process  fill:#1e293b,stroke:#38bdf8,stroke-width:1px,color:#f8fafc
    classDef success  fill:#064e3b,stroke:#34d399,stroke-width:1px,color:#f8fafc
    classDef error    fill:#7f1d1d,stroke:#f87171,stroke-width:1px,color:#f8fafc
    classDef warning  fill:#78350f,stroke:#fbbf24,stroke-width:1px,color:#f8fafc
    classDef entry    fill:#1e3a5f,stroke:#38bdf8,stroke-width:2px,color:#f8fafc
    classDef aux      fill:#334155,stroke:#64748b,stroke-width:1px,color:#f8fafc
    classDef system   fill:#312e81,stroke:#a78bfa,stroke-width:1px,color:#f8fafc
    classDef data     fill:#132d29,stroke:#2dd4bf,stroke-width:1px,color:#f8fafc
    classDef network  fill:#881337,stroke:#fb7185,stroke-width:1px,color:#f8fafc
    classDef accent   fill:#1e1b4b,stroke:#818cf8,stroke-width:1px,color:#f8fafc
    subgraph RawLog["Исходный топик (события по ключам A, B, C)"]
        direction LR
        M1["k:A, v:1"]:::process --> M2["k:B, v:2"] --> M3["k:A, v:3"] --> M4["k:C, v:4"] --> M5["k:A, v:5"]
    end

    subgraph DeletePolicy["Политика: delete (Retention)"]
        D1["Удаление СТАРЫХ СЕГМЕНТОВ<br>по времени (retention.ms) или объему (retention.bytes)"]
    end

    subgraph CompactPolicy["Политика: compact (Log Compaction)"]
        C1["Уплотнение по ключам:<br>k:B -> 2<br>k:C -> 4<br><b>k:A -> 5</b> (дубликаты 1 и 3 удалены)"]
    end

    RawLog -->|"cleanup.policy=delete"| DeletePolicy
    RawLog -->|"cleanup.policy=compact"| CompactPolicy

Retention: классическое удаление по времени и размеру

Политика delete — это классический и наиболее распространенный способ управления размером топиков в Kafka. Топик хранит сообщения в течение заданного окна времени либо до достижения заданного суммарного объема на диске.

Параметры, управляющие retention

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

retention.ms — максимальное время хранения сообщения в топике (по умолчанию 7 дней: 604800000 мс). Сообщения, чей timestamp старше now - retention.ms, становятся кандидатами на удаление.
retention.bytes — максимальный суммарный размер всех сегментов партиции на диске (по умолчанию -1, то есть без ограничения). При превышении этого порога самые старые сегменты удаляются, даже если их возраст меньше retention.ms.
log.retention.check.interval.ms — периодичность, с которой фоновый планировщик Log Manager сканирует директории брокера на соответствие условиям очистки (по умолчанию 5 минут).

Как работает удаление под капотом

Физически операция удаления в Kafka оперирует исключительно целыми файлами-сегментами!

Брокер никогда не пытается «вырезать» отдельные сообщения из середины файла .log — это вызвало бы чудовищный оверхед на дефрагментацию и переписывание сотен гигабайт данных.

flowchart LR
    classDef process  fill:#1e293b,stroke:#38bdf8,stroke-width:1px,color:#f8fafc
    classDef success  fill:#064e3b,stroke:#34d399,stroke-width:1px,color:#f8fafc
    classDef error    fill:#7f1d1d,stroke:#f87171,stroke-width:1px,color:#f8fafc
    classDef warning  fill:#78350f,stroke:#fbbf24,stroke-width:1px,color:#f8fafc
    classDef entry    fill:#1e3a5f,stroke:#38bdf8,stroke-width:2px,color:#f8fafc
    classDef aux      fill:#334155,stroke:#64748b,stroke-width:1px,color:#f8fafc
    classDef system   fill:#312e81,stroke:#a78bfa,stroke-width:1px,color:#f8fafc
    classDef data     fill:#132d29,stroke:#2dd4bf,stroke-width:1px,color:#f8fafc
    classDef network  fill:#881337,stroke:#fb7185,stroke-width:1px,color:#f8fafc
    classDef accent   fill:#1e1b4b,stroke:#818cf8,stroke-width:1px,color:#f8fafc
    subgraph ClosedSegments["Закрытые сегменты (Read-Only)"]
        direction LR
        S1["Сегмент 0<br>(Age: 9 days)"]:::process --> S2["Сегмент 250<br>(Age: 3 days)"]
    end

    subgraph ActiveSegment["Активный сегмент"]
        S3["Сегмент 500<br>(Идет запись!)"]
    end

    ClosedSegments --> ActiveSegment

    Del["Log Manager: retenton.ms = 7 days"] -.->|"unlink(2)"| S1
    S1 -.->|"Удален с диска!"| Trash[("Свободное место")]

Правила работы Log Manager:

Неприкосновенность активного сегмента: Активный сегмент, в который в текущий момент поступают записи от продюсеров, никогда не удаляется, даже если его размер превышает retention.bytes или сообщения внутри него устарели по времени!
Проверка временных меток: Log Manager находит максимальный timestamp среди всех записей сегмента (хранящийся в заголовке .timeindex). Если now - max_segment_timestamp > retention.ms, весь сегмент помечается кандидатом на удаление.
Системный вызов unlink(2): Удаление закрытого сегмента — это тривиальный системный вызов ядра операционной системы unlink(2) (в Java — File.delete()), удаляющий тройку файлов: .log, .index и .timeindex.

Compaction: уплотнение по ключу

Если политика retention отвечает на вопрос: «Как долго хранить временную историю сообщений?», то Log Compaction отвечает на совершенно другой инженерный вопрос: «Какое значение каждого уникального ключа является актуальным на данный момент?».

В компактифицированном топике для каждого уникального ключа гарантированно сохраняется только запись с максимальным offset-ом (последняя по времени записи). Все предшествующие исторические записи с тем же самым ключом признаются устаревшими и безвозвратно удаляются из лога.

Консьюмер, начавший чтение компактифицированного топика с самого начала (offset = 0), гарантированно восстановит полное актуальное состояние системы, прочитав минимально необходимый объем данных.

Зачем нужен compaction

Основные сценарии применения уплотнения лога:

Change Data Capture (CDC): Интеграция с реляционными базами данных через Debezium. Каждое изменение строки (INSERT, UPDATE, DELETE) в PostgreSQL или MySQL публикуется в топик Kafka, где ключом является первичный ключ таблицы (id). Уплотнение лога превращает топик в распределенный материализованный снапшот базы данных!
Хранение эталонных справочников (Reference Data): Таблицы курсов валют, конфигурации фиче-флагов, каталоги категорий товаров.
Системные внутренние топики Kafka: Самый известный пример — топик __consumer_offsets. Каждое сообщение в нем хранит ключ [group_id, topic, partition], а значением является последний зафиксированный offset. Благодаря compaction топик смещений не разрастается до терабайтов, удерживая только последние смещения консьюмеров.

Как работает Log Cleaner под капотом

За процесс компактификации отвечает выделенный пул фоновых потоков брокера — Log Cleaner.

Физическая партиция делится на две логические зоны:

Чистая часть (Clean/Tail): Сегменты, которые уже прошли уплотнение. Здесь каждый ключ встречается ровно один раз.
Грязная часть (Dirty/Head): Сегменты, в которые дописывались новые данные, и где присутствуют повторные записи одних и тех же ключей.

flowchart LR
    classDef process  fill:#1e293b,stroke:#38bdf8,stroke-width:1px,color:#f8fafc
    classDef success  fill:#064e3b,stroke:#34d399,stroke-width:1px,color:#f8fafc
    classDef error    fill:#7f1d1d,stroke:#f87171,stroke-width:1px,color:#f8fafc
    classDef warning  fill:#78350f,stroke:#fbbf24,stroke-width:1px,color:#f8fafc
    classDef entry    fill:#1e3a5f,stroke:#38bdf8,stroke-width:2px,color:#f8fafc
    classDef aux      fill:#334155,stroke:#64748b,stroke-width:1px,color:#f8fafc
    classDef system   fill:#312e81,stroke:#a78bfa,stroke-width:1px,color:#f8fafc
    classDef data     fill:#132d29,stroke:#2dd4bf,stroke-width:1px,color:#f8fafc
    classDef network  fill:#881337,stroke:#fb7185,stroke-width:1px,color:#f8fafc
    classDef accent   fill:#1e1b4b,stroke:#818cf8,stroke-width:1px,color:#f8fafc
    subgraph CleanTail["Чистая часть (Clean / Compacted)"]
        direction LR
        C1["Сегмент 0<br>(Уникальные ключи)"]
    end

    subgraph DirtyHead["Грязная часть (Dirty)"]
        direction LR
        D1["Сегмент 100<br>(Дубликаты)"]
        D2["Активный сегмент 200<br>(Запись продюсера)"]
    end

    CleanTail --> DirtyHead

Алгоритм работы Log Cleaner:

Выбор кандидата: Log Cleaner вычисляет соотношение «грязных» данных к общему размеру закрытых сегментов партиции:
$$\text{Dirty Ratio} = \frac{\text{Bytes in Dirty Segments}}{\text{Bytes in Clean Segments} + \text{Bytes in Dirty Segments}}$$
Очистка запускается только тогда, когда $\text{Dirty Ratio} \ge \textbf{min.cleanable.dirty.ratio}$ (по умолчанию 0.5, то есть 50% грязных данных). Если грязных данных мало, брокер откладывает работу, чтобы не расходовать драгоценные циклы дискового ввода-вывода на переписывание почти чистого лога.
Построение Offset Map (SkimpyOffsetMap): Log Cleaner строит в оперативной памяти компактную хэш-таблицу (используя 128-битный хэш ключа Murmur3 и 64-битный offset). Карта сохраняет отображение: $\text{Key Hash} \to \text{Highest Offset}$.
Линейное слияние (Deduplication Merge): Поток Log Cleaner последовательно читает записи из грязных сегментов. Если offset записи совпадает со значением в Offset Map — запись копируется в новый чистый сегмент. Если offset меньше — запись признается устаревшим дубликатом и пропускается (не попадает в новый файл).
Атомарная подмена сегмента: Новый компактный сегмент завершается, индексы пересчитываются, старые сегменты удаляются через unlink(2), а ссылка на чистый сегмент атомарно обновляется.

flowchart TD
    classDef process  fill:#1e293b,stroke:#38bdf8,stroke-width:1px,color:#f8fafc
    classDef success  fill:#064e3b,stroke:#34d399,stroke-width:1px,color:#f8fafc
    classDef error    fill:#7f1d1d,stroke:#f87171,stroke-width:1px,color:#f8fafc
    classDef warning  fill:#78350f,stroke:#fbbf24,stroke-width:1px,color:#f8fafc
    classDef entry    fill:#1e3a5f,stroke:#38bdf8,stroke-width:2px,color:#f8fafc
    classDef aux      fill:#334155,stroke:#64748b,stroke-width:1px,color:#f8fafc
    classDef system   fill:#312e81,stroke:#a78bfa,stroke-width:1px,color:#f8fafc
    classDef data     fill:#132d29,stroke:#2dd4bf,stroke-width:1px,color:#f8fafc
    classDef network  fill:#881337,stroke:#fb7185,stroke-width:1px,color:#f8fafc
    classDef accent   fill:#1e1b4b,stroke:#818cf8,stroke-width:1px,color:#f8fafc
    subgraph "Сегмент до compaction"
        S1["Offset 0: key=A, val=1"]:::accent
        S2["Offset 1: key=B, val=2"]:::accent
        S3["Offset 2: key=A, val=3"]:::accent
        S4["Offset 3: key=C, val=4"]:::accent
        S5["Offset 4: key=A, val=null<br>tombstone"]:::accent
    end
    subgraph "Offset Map<br>последний offset для каждого ключа"
        M["A: 4, B: 1, C: 3"]:::process
    end
    subgraph "Сегмент после compaction"
        C1["Offset 1: key=B, val=2"]:::accent
        C2["Offset 3: key=C, val=4"]:::accent
        C3["Offset 4: key=A, val=null<br>tombstone"]:::accent
    end
    S1 & S2 & S3 & S4 & S5 -->|"Cleaner"| M
    M -->|"оставляет только последние"| C1 & C2 & C3

Tombstone-записи и `delete.retention.ms`

А как удалить ключ в системе, где лог только дописывается? Для этого продюсер публикует специальную запись — Tombstone («надгробная плита»): сообщение с непустым бизнес-ключом и пустым телом: Value == nil (null).

Однако Tombstone нельзя стереть немедленно в процессе первого же раунда компактификации! Представьте, что консьюмер был временно отключен на 2 часа: если брокер мгновенно удалит и старое значение, и сам Tombstone, проснувшийся консьюмер вообще не узнает, что объект был удален из базы данных!

Параметр delete.retention.ms (по умолчанию 24 часа) определяет, как долго tombstone хранится в логе после того, как он был обработан Log Cleaner и признан последней записью для своего ключа. По истечении этого срока при следующей очистке tombstone будет окончательно удалён, и ключ исчезнет из лога.

sequenceDiagram
    autonumber
    actor App as Продюсер (CDC)
    participant Log as Партиция Kafka
    participant Cleaner as Log Cleaner
    actor Consumer as Отстающий консьюмер

    App->>Log: Запись Tombstone: Key="user-42", Value=nil (offset=100)
    Note over Log,Cleaner: Первый проход Cleaner: старые записи user-42 стёрты,<br>но Tombstone ОСТАЕТСЯ в логе!
    Consumer->>Log: Чтение лога (offset 100): получил Tombstone!<br>Удаляет запись из локальной БД!
    Note over Log,Cleaner: Прошло delete.retention.ms (24 часа)
    Note over Cleaner: Второй проход Cleaner: Tombstone удален с диска без следа!

Ловушка / Gotcha: Воскрешение удаленных сущностей (Zombie Resurrect)

Если консьюмер отстаёт больше, чем на delete.retention.ms, и tombstone уже удалён, при перемотке он увидит старые значения ключа, не узнав, что ключ был удалён. Это приводит к «воскрешению» удалённых записей в состоянии консьюмера. Чтобы избежать этого, необходимо либо гарантировать, что консьюмер не отстаёт на такие длительные периоды, либо использовать компактифицированные топики в сочетании с дополнительным механизмом синхронизации (например, снапшоты в Kafka Streams).

Mechanical Sympathy: влияние на ввод-вывод и страничный кеш

Понимание аппаратной цены каждого механизма очистки критично при проектировании архитектуры:

Retention: почти бесплатно

Удаление сегментов — это чистая метаданная операция операционной системы. Файловая система освобождает экстенты и блоки, не требуя ни единого чтения или записи данных с физического носителя. Страничный кэш ядра Linux автоматически очищается по мере вытеснения ненужных страниц демоном kswapd. Дисковая нагрузка равна нулю.

Compaction: тяжеловесная операция

Log Cleaner — это интенсивная фоновая работа, которая читает старые сегменты и записывает новые. Это конкурирует со штатной нагрузкой брокера (продюсеры, консьюмеры). Ключевые аспекты:

Чтение: холодные сегменты, возможно, вытеснены из Page Cache в момент очистки, что приводит к физическому чтению с диска.
Запись: создание новых сегментов порождает последовательную запись, что эффективно, но добавляет общий объём ввода-вывода (Write Amplification).
Память: Offset Map для каждой партиции может быть значительной (десятки мегабайт), потребляя оперативную память брокера.

Тюнинг log.cleaner.threads (число потоков очистки) и min.cleanable.dirty.ratio позволяет балансировать между свежестью компактификации и нагрузкой на железо.

Сравнение Retention и Compaction

Характеристика	Retention (delete)	Compaction (compact)
Критерий очистки	Время (`retention.ms`) / размер топика (`retention.bytes`)	Последнее значение ключа (`Offset Map`)
Физическая операция	Удаление целых сегментов (`unlink`)	Переписывание сегментов с удалением дубликатов
Порядок сообщений	Сохраняется строго	Сохраняется, но часть сообщений исчезает
Воспроизведение истории	Да, за весь период retention	Нет, только последнее состояние каждого ключа
Накладные расходы CPU/IO	Почти нулевые (только метаданные ФС)	Значительные (I/O чтение/запись, RAM на Offset Map)
Типовые сценарии	Логирование, поток событий, метрики, аналитика	CDC, таблицы состояний, `__consumer_offsets`

Допустимо комбинировать политики: cleanup.policy=compact,delete — топик будет одновременно удалять сообщения по времени/размеру и компактифицировать старые сегменты. Это позволяет ограничить абсолютный объём хранилища, сохраняя преимущества compaction.

Практическое использование в Go: восстановление состояния из компактифицированного топика

Типичный паттерн для stateful-микросервиса на Go: при холодном старте консьюмер читает компактифицированный топик с настройкой auto.offset.reset=earliest и строит в оперативной памяти быструю проекцию состояния (In-Memory K-V Cache):

package main

import (
    "context"
    "fmt"
    "log"
    "strconv"
    "sync"

    "github.com/twmb/franz-go/pkg/kgo"
)

// CurrencyCache представляет потокобезопасный локальный кэш курсов валют
type CurrencyCache struct {
    mu    sync.RWMutex
    rates map[string]float64
}

func NewCurrencyCache() *CurrencyCache {
    return &CurrencyCache{rates: make(map[string]float64)}
}

// Пример: загрузка справочника курсов валют из компактифицированного топика
func loadCurrencyRates(ctx context.Context, client *kgo.Client) (map[string]float64, error) {
    rates := make(map[string]float64)
    for {
        fetches := client.PollFetches(ctx)
        if fetches.IsClientClosed() {
            break
        }
        if ctx.Err() != nil {
            break
        }

        fetches.EachRecord(func(record *kgo.Record) {
            key := string(record.Key)
            // tombstone: ключ удалён из системы
            if record.Value == nil {
                delete(rates, key)
                return
            }
            rate, err := strconv.ParseFloat(string(record.Value), 64)
            if err != nil {
                log.Printf("invalid currency rate format for key %s: %v", key, err)
                return
            }
            rates[key] = rate
        })

        // Коммитить offset имеет смысл только после построения полного состояния,
        // чтобы при перезапуске не перечитывать всё заново.
        client.CommitUncommittedOffsets(ctx)
    }
    return rates, nil
}

Вопросы для собеседования

Собеседование

Вопрос: Почему консьюмер компактифицированного топика при восстановлении состояния должен читать весь лог с начала, а не только последние сообщения?
Ответ: Потому что compaction не даёт гарантии, что для каждого ключа есть запись в последнем сегменте. Ключ мог быть записан однократно в старом сегменте, больше не обновляться, и его значение осталось только там. После прочтения всего лога с начала консьюмер восстанавливает полное актуальное состояние. В Kafka Streams для ускорения используются локальные RocksDB с периодическими снапшотами, чтобы не перечитывать весь лог каждый раз.

Собеседование

Вопрос: Что произойдет, если в компактифицированный топик отправлять сообщения без ключа (key == nil)?
Ответ: Сообщения без ключа не могут быть уплотнены, так как у них нет идентификатора сущности. В старых версиях Kafka такие сообщения удалялись при первой же очистке. В современных версиях брокер не может сопоставить им Offset Map, и они приводят либо к деградации эффективности Log Cleaner, либо сохраняются бесконечно. В компактифицированных топиках ключ сообщения обязателен для каждой записи.

Когда Retention, а когда Compaction: правило выбора

Если поток данных — это непрерывный поток событий (события кликов, логи, метрики), и вам нужна история за последний день/неделю — используйте retention.
Если данные — это изменения сущностей (пользователи, заказы), и вам важно текущее состояние, а не история — используйте compaction.
Если вам нужно и то, и другое (история изменений за период и быстрое восстановление состояния) — комбинируйте: используйте отдельный топик с retention для событий и компактифицированный топик для CDC-снапшотов. Или compact,delete.

Заключение и следующие шаги

Retention и Compaction — два фундаментальных механизма, превращающих Kafka из просто «распределённого лога» в долговременное хранилище, способное обслуживать как операционную аналитику, так и постоянно актуальные таблицы состояний. Понимание их физического устройства — ключ к корректной настройке кластера и предотвращению катастрофических отставаний или потерь данных.

Теперь, когда мы разобрались с хранением, пора переходить к более высокоуровневой обработке данных прямо внутри кластера. В следующей статье мы рассмотрим 9. Kafka Streams — библиотеку, которая строит потоковые топологии поверх тех самых логов, которые мы только что изучили.

Два взгляда на жизненный цикл данных

Retention: классическое удаление по времени и размеру

Параметры, управляющие retention

Как работает удаление под капотом

Compaction: уплотнение по ключу

Зачем нужен compaction

Как работает Log Cleaner под капотом

Tombstone-записи и delete.retention.ms

Mechanical Sympathy: влияние на ввод-вывод и страничный кеш

Retention: почти бесплатно

Compaction: тяжеловесная операция

Сравнение Retention и Compaction

Практическое использование в Go: восстановление состояния из компактифицированного топика

Вопросы для собеседования

Когда Retention, а когда Compaction: правило выбора

Заключение и следующие шаги

Tombstone-записи и `delete.retention.ms`