> Что делать если в Kafka мало места при большом объеме сообщений (Go)

Короткий ответ

Проблема решается на трёх уровнях: retention policy, сжатие и партиционирование. Сначала настройте retention.ms и retention.bytes под реальный SLA потребителей, включите compression.type (zstd или lz4) на producer и broker. Затем проверьте cleanup.policy=compact для keyed-топиков. Если этого мало - добавьте брокеры и перераспределите партиции, либо вынесите холодные данные в S3 через tiered storage.

Подробное объяснение

В Kafka место на диске определяется тремя множителями: объёмом входящего трафика, retention и replication factor. Прежде чем что-то менять, нужно понять, какой из них доминирует. Часто оказывается, что топик хранит данные дольше, чем нужно потребителям, или что compression отключён, и сообщения лежат в plain text.

Первый рычаг - retention. Параметры retention.ms, retention.bytes и segment.bytes задаются на уровне топика и переопределяют broker defaults. Важно помнить, что Kafka удаляет данные только целыми сегментами, поэтому segment.bytes должен быть согласован с retention: слишком крупный сегмент приведёт к перерасходу диска, слишком мелкий - к росту overhead на index-файлы.

Второй рычаг - сжатие. Producer-side compression (compression.type=zstd) снижает сетевой трафик и объём на диске в 3-10 раз для текстовых payload. Broker хранит данные в том виде, в котором получил, поэтому если producer не сжимает, broker не сожмёт за него. Для JSON и логов zstd даёт лучший trade-off между CPU и степенью сжатия, для бинарных protobuf - lz4.

Третий рычаг - compaction. Для топиков, где важен последний state по ключу (CDC, changelog, config), cleanup.policy=compact удаляет старые версии записи. Комбинация compact,delete позволяет и компактить, и ограничивать по времени.

Четвёртый рычаг - топология. Увеличение числа партиций не уменьшает объём данных, но позволяет распределить нагрузку по большему числу брокеров. Если диск упирается в физический лимит одной машины, нужно добавлять брокеры и перераспределять партиции через kafka-reassign-partitions.sh. При этом replication factor напрямую умножает занимаемое место: RF=3 означает три копии каждого байта.

Пятый рычаг - tiered storage (KIP-405). В Kafka 3.6+ появился remote storage: горячие сегменты остаются на локальном диске, холодные уезжают в S3 или HDFS. Это снимает проблему "диск кончился, а данные нужны" без потери consumer offsets.

Наконец, диагностика. Прежде чем крутить параметры, посмотрите kafka-log-dirs.sh --describe и метрики kafka.log:type=Log,name=Size. Это покажет, какие именно топики и партиции съедают диск.

На практике

Начинаю с аудита: какие топики занимают больше всего, какой у них retention и compression. В 80% случаев виновник - топик с дефолтным retention.ms=604800000 (7 дней) и без сжатия, куда пишут debug-логи. Дальше действую по возрастанию риска: сначала включаю zstd на producer, потом сокращаю retention, потом добавляю брокеры.

Важный нюанс: изменение compression.type на лету не пересжимает уже записанные сегменты. Экономия появится только на новых данных, старые будут занимать место до истечения retention. Если нужно освободить место срочно - можно форсировать roll сегментов через kafka-topics.sh --alter --config segment.ms=..., но это создаёт нагрузку на I/O.

Для Go-сервисов использую segmentio/kafka-go или confluent-kafka-go. В kafka-go сжатие задаётся через Writer.Compression, в confluent - через compression.codec в конфиге. На стороне Java - compression.type в ProducerConfig.

При tiered storage нужно учитывать, что consumer fetch с remote storage имеет более высокий latency. Для latency-sensitive потребителей лучше держать local.retention.ms достаточным, чтобы горячие чтения не уходили в S3.

Пример кода

Настройка producer в Go с zstd и ограничением размера батча:

GO
package main

import (
    "context"
    "log"
    "time"

    "github.com/segmentio/kafka-go"
    "github.com/segmentio/kafka-go/compress"
)

func main() {
    w := &kafka.Writer{
        Addr:         kafka.TCP("broker1:9092", "broker2:9092"),
        Topic:        "events",
        Balancer:     &kafka.Hash{},
        Compression:  compress.Zstd,
        BatchSize:    1000,
        BatchTimeout: 100 * time.Millisecond,
        RequiredAcks: kafka.RequireAll,
    }
    defer w.Close()

    err := w.WriteMessages(context.Background(),
        kafka.Message{Key: []byte("user-1"), Value: []byte(`{"event":"click"}`)},
    )
    if err != nil {
        log.Fatalf("write failed: %v", err)
    }
}

Настройка топика через CLI:

BASH
kafka-configs.sh --bootstrap-server broker1:9092 \
  --entity-type topics --entity-name events --alter \
  --add-config 'retention.ms=86400000,retention.bytes=107374182400,segment.bytes=1073741824,compression.type=zstd,cleanup.policy=delete'

Как отвечать на собеседовании

Структурируйте ответ как диагностику, а не как список трюков. Сначала уточните: какой retention, есть ли compression, сколько партиций и какой RF, есть ли tiered storage. Потом предложите решения в порядке возрастания стоимости: сжатие, retention, compaction, добавление брокеров, tiered storage. Упомяните, что удаление идёт по сегментам, и что replication factor умножает занимаемое место. Покажите, что понимаете trade-off между latency и экономией диска.

Что проверяет интервьюер

Понимание внутреннего устройства Kafka: сегменты, retention, compaction, replication. Умение диагностировать проблему, а не сразу кидаться менять конфиги. Знание, что compression задаётся на producer и не пересжимает старые данные. Понимание, что горизонтальное масштабирование - не единственное решение, и что tiered storage существует. Способность объяснить trade-off между стоимостью диска, latency и сложностью эксплуатации.

Типичные ошибки

Увеличение числа партиций без добавления брокеров - это не уменьшает объём данных, а только размазывает его по тем же дискам. Включение compression на broker в надежде, что он пересожмёт старые сегменты. Забывание про replication factor при расчёте места. Установка retention.bytes меньше, чем размер одного сегмента, что приводит к удалению активного сегмента. Игнорирование consumer lag: сокращение retention без проверки, что потребители успевают читать, приводит к потере данных.