Tiered Storage в Kafka: горячие данные на диске, холодные в S3-совместимом хранилище (KIP-405 + SeaweedFS)

Retention в Kafka упирается не в пропускную способность, а в диск. Бизнесу нужен год истории, брокеры держат неделю. Ниже разберём, как это решает Tiered Storage (KIP-405): что происходит с сегментом, как читаются старые офсеты, где схема ограничена. Затем проверим её на одном брокере с S3-совместимым хранилищем.

Важная оговорка о статусе текста. Всё, что сказано о поведении стенда, ниже сформулировано как гипотеза из документации. Подтверждать или опровергать её должны блоки output под скриптами, которые подставляются из реального прогона. Если вывод разойдётся с гипотезой, верить нужно выводу. Пока в блоках нет вывода, ни одно утверждение о стенде нельзя считать проверенным.

Разбор: что сказано в исходном посте

Пост на dev.to описывает идею так: свежие данные лежат на локальном диске брокера, закрытые и состарившиеся сегменты Kafka выгружает в S3 и удаляет локальную копию. Топик и офсеты остаются прежними, консьюмер читает шестимесячные данные, просто медленнее. Две пары настроек задают всю схему: local.retention.* определяет горячее окно на диске, retention.* — полную историю. В посте есть конфиги под EKS с IRSA и Aiven-плагином, оценка стоимости и отдельная часть про Databricks. Часть про Databricks и цены мы не берём: она не по теме и не воспроизводится на стенде. Автор поста также подчёркивает, что tiered storage не бэкап. Мы не принимаем это на веру, а проверим в задании 3.

Конфиги из поста нельзя копировать как есть. В нём remote.storage.manager.class.name, а в README плагина Aiven параметры называются remote.log.storage.manager.class.name и remote.log.storage.manager.class.path. Класс S3-бэкенда в посте — S3StorageBackend, но в README приведён только пример для GCS (...storage.gcs.GcsStorage). Всё, что ниже, сверено с README и KIP-405, а непроверенное помечено.

Архитектура KIP-405

В KIP три сущности.

• RemoteLogManager (RLM) — внутренний компонент брокера, не публичный API. Он получает события о смене лидерства и остановке или удалении партиций и делегирует копирование, чтение и удаление сегментов плагинам.
• RemoteStorageManager (RSM) — интерфейс хранилища сегментов и индексов. Реализации (HDFS, S3) по замыслу KIP живут вне репозитория Kafka, как коннекторы. В самой Kafka есть фреймворк, но нет готового S3-плагина.
• RemoteLogMetadataManager (RLMM) — интерфейс метаданных об удалённых сегментах со строго согласованной семантикой. По умолчанию это реализация на внутреннем топике (TopicBasedRemoteLogMetadataManager).

Метаданные вынесли из удалённого хранилища сознательно. В KIP объяснено: при метаданных в S3 нужны LIST-операции, строгая согласованность (GET после DELETE) и лишние расходы на частые LIST.

Для каждой партиции RLM создаёт задачи. Лидерская задача ищет закрытые сегменты и копирует их вместе с offset/time/transaction/producer-snapshot индексами и кэшем leader epoch. Она же обслуживает fetch старых данных. Фолловерская задача отслеживает удалённые сегменты через RLMM. RLM держит ограниченный кэш индексов удалённых сегментов в каталоге remote-log-index-cache внутри log dir. Размер задаётся remote.log.index.file.cache.total.size.mb.

В KIP заданы границы офсетов лидера: Lz ≥ Ly ≥ Lx и Ly ≥ Ry ≥ Rx, где Lx — начало локального лога, Lz — его конец, Ly — LSO, Rx и Ry — начало и конец удалённого лога. Конец удалённого лога не уходит за LSO.

Жизненный цикл сегмента и две пары retention

1. Продюсер пишет в активный сегмент, ничего не меняется (acks, репликация как раньше).
2. Сегмент закрывается (roll). По KIP копируются сегменты, у которых последний офсет меньше LSO. README плагина Aiven: загрузка идёт последовательно, от старого сегмента к новому, без активного.
3. Копия лежит в хранилище, метаданные опубликованы через RLMM.
4. Локальная копия удаляется по local.retention.*. В KIP сказано: локальные логи не чистятся, пока сегменты не скопированы, даже если срок или размер retention уже достигнут.
5. Из удалённого хранилища сегмент удаляется по retention.*. Эту часть мы на стенде не проверяем.

Из этого следуют практические вопросы. Это выводы из KIP и README, на стенде они не измерялись.

• Соотношение local.retention.* и retention.*. По смыслу из поста полная история включает локальную, но требование к значениям в предоставленных материалах не подтверждено. Как брокер валидирует сочетание параметров, смотрите в документации Kafka для своей версии.
• Отстающая выгрузка. Раз KIP не разрешает чистить локальные логи до копирования, при медленном S3 или жёстком rate limit плагина локальные данные не удаляются. Диск при этом может расти сверх local.retention. Это прямое следствие KIP, но мы его не замеряли.
• Активный сегмент. README говорит, что загрузка идёт без активного сегмента. Значит, хвост размером до segment.bytes на партицию остаётся локально. Из этого следует, что при большом segment.bytes и редком roll свежие данные долго не окажутся в S3. Это наш вывод, не цитата.

Фолловеры реплицируют только локальные данные лидера. Новая реплика получает OFFSET_MOVED_TO_TIERED_STORAGE, узнаёт earliest local offset и достраивает leader epoch и producer snapshot из удалённого хранилища. В мотивации KIP сказано, что время восстановления и ребалансировки пропорционально объёму данных, которые брокер хранит локально. С tiered storage этот локальный объём меньше.

Путь чтения

Запрос свежего офсета обслуживается с диска и из page cache, как раньше. Запрос старого офсета лидер находит через RLMM, а плагин читает нужный диапазон из объекта и отдаёт брокеру. Консьюмер разницы по офсетам не видит. Что меняется по задержке, покажет задание 2.

По README плагина Aiven:

• Сегмент режется на чанки (рекомендуемый размер — 4 MiB, rsm.config.chunk.size). Чанки позволяют делать ranged-запросы при сжатии и шифровании, не скачивая весь объект.
• Есть кэш чанков: в памяти и на диске (DiskChunkCache). В README сказано, что в большинстве случаев in-memory вариант не подходит для продакшена. Есть и prefetch следующих чанков, но только в границах сегмента.
• Таймаут удалённого fetch по умолчанию 500 мс. Если его не хватает, задача отменяется. Параметр remote.fetch.max.wait.ms увеличивает таймаут ценой роста задержки.
• Выгрузка использует multipart upload. README рекомендует lifecycle-правило на бакете для abort незавершённых multipart-загрузок.
• Выгрузку можно ограничить по скорости (upload.rate.limit.bytes.per.second). По умолчанию пул из 10 потоков.

Альтернативы

Ограничения

• Статус. По шапке KIP-405, фича объявлена production-ready с Kafka 3.9. Ограничения GA перечислены в отдельных release notes, на которые ссылается KIP. Страницу документации kafka.apache.org/documentation/#tiered_storage при подготовке статьи прочитать не удалось, поэтому сверяйтесь с ней сами.
• Compacted-топики не поддерживаются. Топик с remote.storage.enable=true нельзя перевести с политики delete на compact.
• JBOD с tiered storage не поддерживается (раздел Non-Goals в KIP).
• Не заменяет ETL. Это сказано в KIP прямо.
• Не бэкап — гипотеза, не установленный факт. В KIP сказано только, что RLM получает события stop/delete партиций. Описания того, что именно происходит с объектами при удалении топика, в наших материалах нет. Что остаётся в бакете после удаления топика, показывает задание 3.
• Ограничения на отключение remote.storage.enable на живом топике в предоставленных материалах не подтверждены. Проверьте их в документации своей версии до включения на продакшене.
• Совместимость плагина Aiven с Kafka 4.3.2 в материалах не подтверждена. Стенд ниже её проверяет, но не гарантирует.

Практика

Стенд: один брокер Kafka 4.3.2 в KRaft (combined mode), SeaweedFS как S3-совместимое хранилище, плагин Aiven. Плагин скрипт скачивает из последнего релиза GitHub, не угадывая версию. Имена артефактов (core-<version>.tgz, s3-<version>.tgz) взяты из README. Класс S3-бэкенда в README не приведён, поэтому скрипт ищет io.aiven.kafka.tieredstorage.storage.s3.S3Storage в jar по аналогии с GCS и падает, если его нет. Имена параметров storage.s3.endpoint.url и storage.s3.path.style.access.enabled также не подтверждены документацией из входных данных. Если брокер не стартует, начните сверку с docs/ в репозитории плагина.

Файл compose.yaml:

services:
  seaweed:
    image: chrislusf/seaweedfs:4.48
    command: server -s3 -dir=/data
    mem_limit: 512m
    ports: ["127.0.0.1:8333:8333"]
  kafka:
    image: apache/kafka:4.3.2
    mem_limit: 2g
    depends_on: [seaweed]
    ports: ["127.0.0.1:9092:9092"]
    tmpfs: ["/cache:mode=1777"]
    volumes: ["./plugin:/plugin:ro"]
    environment:
      - KAFKA_NODE_ID=1
      - KAFKA_PROCESS_ROLES=broker,controller
      - KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092
      - KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093
      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
      - KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1
      - KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1
      - KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0
      - KAFKA_LOG_DIRS=/var/lib/kafka/data
      - KAFKA_HEAP_OPTS=-Xms768m -Xmx768m
      - KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS=5000
      - KAFKA_REMOTE_LOG_MANAGER_TASK_INTERVAL_MS=5000
      - KAFKA_REMOTE_FETCH_MAX_WAIT_MS=5000
      - KAFKA_REMOTE_LOG_STORAGE_SYSTEM_ENABLE=true
      - KAFKA_REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME=PLAINTEXT
      - KAFKA_RLMM_CONFIG_REMOTE_LOG_METADATA_TOPIC_REPLICATION_FACTOR=1
      - KAFKA_RLMM_CONFIG_REMOTE_LOG_METADATA_TOPIC_MIN_ISR=1
      - KAFKA_REMOTE_LOG_STORAGE_MANAGER_CLASS_PATH=/plugin/core/*:/plugin/s3/*
      - KAFKA_REMOTE_LOG_STORAGE_MANAGER_CLASS_NAME=io.aiven.kafka.tieredstorage.RemoteStorageManager
      - KAFKA_RSM_CONFIG_CHUNK_SIZE=4194304
      - KAFKA_RSM_CONFIG_STORAGE_BACKEND_CLASS=io.aiven.kafka.tieredstorage.storage.s3.S3Storage
      - KAFKA_RSM_CONFIG_STORAGE_S3_BUCKET_NAME=kafka-tiered
      - KAFKA_RSM_CONFIG_STORAGE_S3_REGION=us-east-1
      - KAFKA_RSM_CONFIG_STORAGE_S3_ENDPOINT_URL=http://seaweed:8333
      - KAFKA_RSM_CONFIG_STORAGE_S3_PATH_STYLE_ACCESS_ENABLED=true
      - KAFKA_RSM_CONFIG_FETCH_CHUNK_CACHE_CLASS=io.aiven.kafka.tieredstorage.fetch.cache.DiskChunkCache
      - KAFKA_RSM_CONFIG_FETCH_CHUNK_CACHE_PATH=/cache
      - KAFKA_RSM_CONFIG_FETCH_CHUNK_CACHE_SIZE=268435456
      - KAFKA_RSM_CONFIG_FETCH_CHUNK_CACHE_RETENTION_MS=600000
      - AWS_ACCESS_KEY_ID=x
      - AWS_SECRET_ACCESS_KEY=x

Учётные данные AWS берутся из окружения (как и советует пост: ключей в конфиге нет). SeaweedFS без настроенных identities, по нашему предположению, авторизацию не требует, поэтому значения произвольные. Это тоже проверяет прогон.

Файл requirements.txt:

confluent-kafka==2.6.1
boto3==1.35.99

Файл fetch_plugin.py:

import glob, io, json, pathlib, sys, tarfile, urllib.request, zipfile

API = "https://api.github.com/repos/Aiven-Open/tiered-storage-for-apache-kafka/releases/latest"
rel = json.load(urllib.request.urlopen(API))
print("release:", rel["tag_name"])
for part in ("core", "s3"):
    dst = pathlib.Path("plugin") / part
    dst.mkdir(parents=True, exist_ok=True)
    asset = next(a for a in rel["assets"]
                 if a["name"].startswith(part + "-") and a["name"].endswith(".tgz"))
    data = urllib.request.urlopen(asset["browser_download_url"]).read()
    with tarfile.open(fileobj=io.BytesIO(data)) as t:
        for m in t.getmembers():
            if m.isfile() and m.name.endswith(".jar"):
                (dst / pathlib.Path(m.name).name).write_bytes(t.extractfile(m).read())
cls = "io/aiven/kafka/tieredstorage/storage/s3/S3Storage.class"
ok = any(cls in zipfile.ZipFile(j).namelist() for j in glob.glob("plugin/s3/*.jar"))
sys.exit(0 if ok else "S3Storage не найден в jar: сверьте имя backend-класса с README плагина")

Общие функции: создание топика, запись, листинг локальных сегментов и объектов бакета.

Файл common.py:

import os, subprocess, time
import boto3
from confluent_kafka import Producer
from confluent_kafka.admin import AdminClient, NewTopic

BS, BUCKET = "127.0.0.1:9092", "kafka-tiered"

def s3():
    return boto3.client("s3", endpoint_url="http://127.0.0.1:8333", region_name="us-east-1",
                        aws_access_key_id="x", aws_secret_access_key="x")

def ensure_bucket():
    for _ in range(30):
        try:
            s3().create_bucket(Bucket=BUCKET)
            return
        except Exception:
            time.sleep(2)

def objects():
    r = s3().list_objects_v2(Bucket=BUCKET)
    return [(o["Key"], o["Size"]) for o in r.get("Contents", [])]

def _sh(cmd):
    r = subprocess.run(["docker", "compose", "exec", "-T", "kafka", "sh", "-c", cmd],
                       capture_output=True, text=True)
    return r.stdout

def local_files(topic):
    out = _sh(f"ls -l /var/lib/kafka/data/{topic}-0 | grep -E '[.]log$'")
    return [l.split()[-1] for l in out.splitlines()]

def create_topic(name, cfg=None):
    conf = {"remote.storage.enable": "true", "segment.bytes": "1048576",
            "local.retention.ms": "30000", "retention.ms": "86400000"}
    conf.update(cfg or {})
    admin = AdminClient({"bootstrap.servers": BS})
    admin.create_topics([NewTopic(name, 1, 1, config=conf)])[name].result()

def produce(name, mb):
    p = Producer({"bootstrap.servers": BS, "linger.ms": 20})
    for _ in range(mb * 1000):
        p.produce(name, os.urandom(1000))
        p.poll(0)
    p.flush()

def wait_tiered(name, max_local=2, timeout=120):
    end = time.time() + timeout
    while time.time() < end:
        if len(local_files(name)) <= max_local and objects():
            return
        time.sleep(5)
    print(f"{name}: таймаут ожидания выгрузки в remote storage ({timeout} с)")

def report(name):
    kb = _sh(f"du -sk /var/lib/kafka/data/{name}-0").split()[0]
    objs = objects()
    print(f"{name}: локальных .log={len(local_files(name))}, каталог партиции={kb} KiB")
    print(f"бакет: объектов={len(objs)}, всего={sum(s for _, s in objs) // 1024} KiB")

Файл producer.py:

from common import *

ensure_bucket()
create_topic("tiered")
produce("tiered", 40)
wait_tiered("tiered")
report("tiered")

Файл consumer.py:

import time, uuid
from confluent_kafka import Consumer, TopicPartition
from common import BS

def read(topic, back=None):
    """Читает с начала (back=None) или последние back записей. Возвращает (первая, всего, n)."""
    c = Consumer({"bootstrap.servers": BS, "group.id": str(uuid.uuid4()),
                  "enable.auto.commit": False})
    lo, hi = c.get_watermark_offsets(TopicPartition(topic, 0), timeout=10)
    start = lo if back is None else hi - back
    c.assign([TopicPartition(topic, 0, start)])
    t0, first, n, end = time.time(), None, 0, time.time() + 120
    while n < hi - start and time.time() < end:
        m = c.poll(1)
        if m is None or m.error():
            continue
        first = first if first is not None else time.time() - t0
        n += 1
    c.close()
    return first, time.time() - t0, n

if __name__ == "__main__":
    for label, back in (("с offset 0 (remote)", None), ("последние 1000 (local)", 1000)):
        f, t, n = read("tiered", back)
        print(f"{label}: записей={n}, первая выборка={f:.3f}с, всего={t:.3f}с")

Файл run.sh:

#!/usr/bin/env bash
set -euo pipefail
cd "$(dirname "$0")"
python3 fetch_plugin.py
docker compose up -d
for i in $(seq 1 60); do
  if curl -s -o /dev/null http://127.0.0.1:8333 && \
     docker compose exec -T kafka /opt/kafka/bin/kafka-topics.sh \
       --bootstrap-server localhost:9092 --list >/dev/null 2>&1; then break; fi
  sleep 3
done
python3 -m venv .venv && .venv/bin/pip install -q -r requirements.txt
.venv/bin/python producer.py
.venv/bin/python consumer.py

Вывод run.sh:

release: v1.1.1
 Network sat-kafka-595_default  Creating
 Network sat-kafka-595_default  Created
 Container sat-kafka-595-seaweed-1  Creating
 Container sat-kafka-595-seaweed-1  Created
 Container sat-kafka-595-kafka-1  Creating
 Container sat-kafka-595-kafka-1  Created
 Container sat-kafka-595-seaweed-1  Starting
 Container sat-kafka-595-seaweed-1  Started
 Container sat-kafka-595-kafka-1  Starting
 Container sat-kafka-595-kafka-1  Started
tiered: таймаут ожидания выгрузки в remote storage (120 с)
tiered: локальных .log=41, каталог партиции=39936 KiB
бакет: объектов=0, всего=0 KiB
с offset 0 (remote): записей=40000, первая выборка=0.008с, всего=0.276с
последние 1000 (local): записей=1000, первая выборка=0.009с, всего=0.029с

Как читать этот вывод. Ниже описано, чего ждать по документации. Это ожидания, а не результат: сравнивайте их с тем, что напечатал прогон.

1. Локальные .log и объекты бакета. По схеме KIP после wait_tiered локально должно остаться мало сегментов, а объекты (сегменты, индексы, манифесты) должны лежать в бакете. Размер каталога партиции стоит сравнить с 40 МБ записанных данных. Если локальных файлов много или бакет пуст, wait_tiered мог просто выйти по таймауту: он не бросает ошибку, а report всё равно напечатает состояние. Тогда выгрузка не произошла или не успела, и причину ищите в логах брокера.
2. Чтение с offset 0. Консьюмер должен получить все 40 000 записей. Сколько пришло на самом деле, показывает поле записей=. Если оно меньше, чтение могло упереться в лимит 120 с или в таймаут удалённого fetch. Время чтения с offset 0 стоит сравнить со временем чтения последних 1000 записей с диска. Мы не предсказываем цифры.
3. Если брокер не стартовал, смотрите docker compose logs kafka. Типичные причины: неверное имя класса RSM или backend, несовместимость плагина с Kafka 4.3.2, неверное имя параметра endpoint.

Задания

Задание 1. Подберите настройки топика так, чтобы после записи 50 МБ локально оставалось не больше двух сегментов, а остальное лежало в бакете. Покажите это листингом log.dirs и содержимым бакета. Объясните, почему активный сегмент не выгружается.

from common import *

create_topic("task1", {"local.retention.bytes": "1"})
produce("task1", 50)
wait_tiered("task1")
report("task1")

task1: таймаут ожидания выгрузки в remote storage (120 с)
task1: локальных .log=52, каталог партиции=49932 KiB
бакет: объектов=0, всего=0 KiB

Задание 2. Измерьте задержку первой выборки при чтении с offset 0 и с недавнего офсета. Повторите чтение с offset 0 несколько раз и сравните проходы. Какую роль мог сыграть кэш чанков?

from consumer import read

for i in range(3):
    f, t, n = read("tiered", None)
    print(f"offset 0, проход {i + 1}: первая={f:.3f}с, всего={t:.3f}с, n={n}")
f, t, n = read("tiered", 1000)
print(f"последние 1000: первая={f:.3f}с, всего={t:.3f}с, n={n}")

offset 0, проход 1: первая=0.007с, всего=0.165с, n=40000
offset 0, проход 2: первая=0.004с, всего=0.134с, n=40000
offset 0, проход 3: первая=0.010с, всего=0.139с, n=40000
последние 1000: первая=0.011с, всего=0.016с, n=1000

Задание 3. Удалите топик и проверьте содержимое бакета. Затем опишите, как получить настоящий бэкап, и какие из вариантов (версионирование бакета, репликация бакета, MirrorMaker 2) защищают от случайного удаления топика.

import time
from confluent_kafka.admin import AdminClient
from common import *

base = {k for k, _ in objects()}
create_topic("task3")
produce("task3", 5)
wait_tiered("task3", timeout=90)
mine = {k for k, _ in objects()} - base
print("объектов топика в бакете:", len(mine))
admin = AdminClient({"bootstrap.servers": BS})
admin.delete_topics(["task3"])["task3"].result()
left = mine
for _ in range(24):
    left = {k for k, _ in objects()} & mine
    if not left:
        break
    time.sleep(5)
print("после удаления топика осталось объектов:", len(left))

task3: таймаут ожидания выгрузки в remote storage (90 с)
объектов топика в бакете: 0
после удаления топика осталось объектов: 0

Источники

• KIP-405: Kafka Tiered Storage
• Tiered Storage for Apache Kafka (плагин Aiven, README)
• Исходный пост на dev.to

Если вы эксплуатируете кластер, где диски, ребалансировка и retention уже стали узким местом, разбор tiered storage вместе с мониторингом, настройкой брокеров и планированием ёмкости есть в курсе «Администрирование кластера Kafka».