Что такое топик (topic) в Kafka

Главное отличие топика Kafka от классических очередей сообщений (RabbitMQ, ActiveMQ):

🟢 Junior Level

30-секундный ответ

Топик (Topic) в Apache Kafka — это именованная логическая категория (канал), представляющая собой распределённый, неизменяемый журнал дозаписи (Immutable Append-Only Commit Log), в который продюсеры публикуют события, а консьюмеры их вычитывают.

Главное отличие топика Kafka от классических очередей сообщений (RabbitMQ, ActiveMQ):

В традиционной очереди: сообщение удаляется брокером сразу после того, как один получатель подтвердил его обработку (destructive read).
В топике Kafka: события НЕ удаляются при чтении. Они сохраняются на диск на заданный период времени (retention policy, например 7 дней). Множество независимых систем и микросервисов могут читать один и тот же топик параллельно, со своей собственной скоростью и даже перечитывать старые данные заново с любого момента времени.

Наглядная аналогия из жизни

Топик — это ежедневная печатная газета или журнал в библиотеке:

Издательство (Продюсер) каждый день печатает новые выпуски и складывает их в архивную подшивку.
Посетители библиотеки (Консьюмеры) приходят в читальный зал. Один читает свежий выпуск за сегодня, другой перечитывает статьи за прошлый вторник, а третий делает полную историческую выписку за весь год.
Чтение газеты одним читателем никак не уничтожает страницы — журнал остаётся лежать на полке для других людей, пока библиотекарь не спишет старые выпуски по регламенту хранения (retention).

Базовый пример отправки и чтения в Spring Boot 3

@Service
public class OrderEventService {

    private final KafkaTemplate<String, String> kafkaTemplate;

    public OrderEventService(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    // Отправка события в топик "orders" с ключом заказа
    public void publishOrderCreated(String orderId, String payload) {
        kafkaTemplate.send("orders", orderId, payload);
    }

    // Независимое чтение из топика группой микросервисов
    @KafkaListener(topics = "orders", groupId = "billing-service-group")
    public void consumeForBilling(String message) {
        // Обработка платежа
    }

    @KafkaListener(topics = "orders", groupId = "notifications-service-group")
    public void consumeForNotifications(String message) {
        // Отправка SMS клиенту (читает тот же топик параллельно)
    }
}

# Создание топика через CLI: 3 партиции для параллелизма, фактор репликации 3
kafka-topics.sh --bootstrap-server localhost:9092 \
  --create \
  --topic orders \
  --partitions 3 \
  --replication-factor 3

🟡 Middle Level

Логическая и физическая архитектура топика

Топик — это лишь логическая абстракция. Физически на серверах-брокерах топик дробится на партиции (Partitions):

                            [ Логический Топик: orders ]
                                         │
        ┌────────────────────────────────┼────────────────────────────────┐
        ▼                                ▼                                ▼
  [ Партиция 0 ]                   [ Партиция 1 ]                   [ Партиция 2 ]
  (Брокер 1, Лидер)                (Брокер 2, Лидер)                (Брокер 3, Лидер)
        │                                │                                │
  Файлы на диске:                  Файлы на диске:                  Файлы на диске:
  orders-0/                        orders-1/                        orders-2/
   ├── 00000.log (Данные)           ├── 00000.log                    ├── 00000.log
   ├── 00000.index (Смещение)       ├── 00000.index                  ├── 00000.index
   └── 00000.timeindex (Время)      └── 00000.timeindex              └── 00000.timeindex

Папка на диске: на файловой системе брокера создаётся каталог <topic_name>-<partition_id>.
Сегменты журнала (.log): данные пишутся в активный файл сегмента. При достижении segment.bytes (по умолчанию 1 ГБ) или segment.ms (по умолчанию 7 дней) сегмент закрывается (sealed) и создаётся новый.
Индексы (.index, .timeindex): разреженные индексы (Memory-Mapped Files), позволяющие мгновенно находить физическое смещение байтов на диске по порядковому номеру (offset) или таймстемпу события без полного сканирования файла.

Политики очистки журнала: `delete` vs `compact`

Поведение удаления данных определяется конфигурацией cleanup.policy:

Политика	Механизм	Назначение / Use Case
`delete` (default)	Старые сегменты безвозвратно удаляются по времени (`retention.ms`) или по общему размеру топика (`retention.bytes`).	Событийные потоки: аудит, метрики, кликстрим, заказы.
`compact`	Фоновый процесс `LogCleaner` удаляет устаревшие дубликаты записей с одинаковым ключом, сохраняя только последнее состояние каждого ключа.	Хранение агрегатов, профили пользователей, кэш-таблицы (Kafka Streams KTable).

Гарантии порядка сообщений

Внутри одной партиции: гарантируется строгий порядок FIFO (First-In, First-Out). Каждое сообщение получает монотонно возрастающий номер — Offset.
В масштабе всего топика (между разными партициями): глобального порядка НЕ существует. Если в топике 3 партиции, события будут считываться параллельно и могут перемешиваться.
Вывод: для обеспечения порядка связанных бизнес-событий (например, событий одного заказа order-123) продюсер обязан указывать одинаковый Message Key. Все сообщения с одним ключом гарантированно попадут в одну и ту же партицию.

🔴 Senior Level

Zero-Copy и операционная система: почему Kafka пишет на диск быстрее, чем другие в память

Распространённый вопрос на собеседованиях: «Как Kafka умудряется обрабатывать миллионы сообщений в секунду, если она каждое сообщение пишет на медленный диск?»

Секрет кроется в архитектуре ядра Linux и отказе от кэширования внутри JVM:

Последовательный ввод-вывод (Sequential I/O): Kafka только дописывает данные в конец файла (Append-Only). Скорость последовательной записи на современный NVMe SSD достигает 3–5 ГБ/с, что сравнимо со скоростью оперативной памяти, исключая случайные позиционирования головки (Random I/O).
Page Cache ядра Linux: брокер Kafka не хранит сообщения в Java Heap (что исключает накладные расходы на сборку мусора GC и оверхед заголовков объектов JVM). Все записанные данные остаются в системном кэше страниц ОС.

Системный вызов sendfile() (Zero-Copy Transfer): При передаче данных консьюмеру Kafka не копирует байты в пользовательское пространство Java. Ядро передаёт данные напрямую из кэша страниц диска в сетевой сокет:

Классический подход: Диск ──► Ядро (PageCache) ──► JVM Heap ──► Буфер сокета ──► NIC (4 копирования)
Zero-Copy (sendfile): Диск ──► Ядро (PageCache) ──────────────────────────────► Сетевая карта (NIC DMA)

Log Compaction, Tombstones и риск «Воскрешения данных»

При использовании cleanup.policy=compact:

Если клиент хочет удалить запись с ключом userId=42, продюсер отправляет в топик специальную запись-надгробие — Tombstone: сообщение с ключом key="42" и пустым телом value=null.
Фоновый поток LogCleaner при проходе сжимает сегменты, удаляет все старые версии значения ключа 42 и временно оставляет сам маркер Tombstone.
Опасная ошибка (Data Resurrection): параметр delete.retention.ms (по умолчанию 24 часа) определяет, как долго Tombstone физически хранится на диске. Если медленный или упавший консьюмер был выключен дольше, чем delete.retention.ms, при повторном запуске он не увидит Tombstone и восстановит старое состояние пользователя из предыдущих бэкапов!

Контракт надёжности: Триада `acks=all`, `min.insync.replicas` и RF

Для предотвращения потери данных в критических финансовых топиках используется жесткая связка настроек:

replication.factor = 3 (3 физические копии партиции на разных серверах).
min.insync.replicas = 2 (минимум 2 живые реплики должны подтвердить запись).
acks = all (или -1) на стороне Producer.

Пограничный сценарий (Edge Case): Если в кластере из 3 брокеров два узла выходят из строя, в строю остаётся только лидер.

Продюсер с acks=all пытается отправить сообщение.
Лидер видит, что размер пула синхронных реплик (ISR) упал до 1, что меньше min.insync.replicas: 2.
Лидер отклоняет запись и выбрасывает продюсеру NotEnoughReplicasException. Это защищает систему от потери данных при последующем перезапуске кластера.

4 Tricky Questions

1. Почему в Kafka категорически запрещено УМЕНЬШАТЬ количество партиций у существующего топика?

Ответ:

В Kafka данные внутри партиции являются неизменяемым последовательным логом, упорядоченным по смещениям (offsets).
Если бы брокер позволил уменьшить число партиций (например, с 4 до 2), возникла бы неразрешимая архитектурная дилемма: что делать с данными из партиций 2 и 3?
- Слить их в партиции 0 и 1 невозможно: нарушится строгая монотонная последовательность смещений (offsets), возникнут коллизии и дубликаты индексов.
- Удалить эти данные нельзя: это приведет к катастрофической потере данных.
Поэтому утилита kafka-topics.sh --alter умеет только увеличивать количество партиций. Если уменьшение всё же необходимо, единственный путь — создать новый топик и запустить миграцию данных через Kafka Connect или MirrorMaker.

2. Что произойдёт с порядком сообщений в системе, если администратор увеличит число партиций с 3 до 6 на боевом топике, куда продюсеры пишут данные с ключом?

Ответ: Порядок обработки сообщений по конкретному бизнес-ключу будет фатально нарушен! По умолчанию продюсер Kafka вычисляет номер партиции по формуле: \(\text{partition} = \text{murmur2}(\text{key}) \pmod{\text{totalPartitions}}\)

При totalPartitions = 3 события по заказу order-99 стабильно попадали, например, в партицию 1.
Как только администратор увеличил число партиций до 6, делитель формулы изменился: новые события по тому же заказу order-99 начнут попадать в партицию 4!
В результате события OrderCancelled в партиции 4 могут быть обработаны консьюмером раньше, чем событие OrderCreated в партиции 1.
Вывод: число партиций на ключевых топиках необходимо рассчитывать и закладывать с многократным запасом на этапе проектирования системы.

3. Почему Kafka не кэширует сообщения в оперативной памяти Java-процесса (JVM Heap), а целиком полагается на Page Cache ядра Linux?

Ответ: Это гениальное архитектурное решение Джея Крепса по трём причинам:

Оверхед объектов Java: каждый объект в Java имеет служебный заголовок (Mark Word + Klass Word) и выравнивание памяти. Сообщение размером 100 байт в JVM Heap может занимать до 300–400 байт. Page Cache хранит сырые сжатые бинарные байты без оверхеда.
Паузы сборщика мусора (GC STW): кэширование десятков гигабайт сообщений в Heap привело бы к колоссальным паузам Full GC, парализующим брокер.
Сохранность при рестарте брокера: если процесс Kafka аварийно завершится (crash / OOM / kill -9) или перезапустится при обновлении, JVM Heap мгновенно уничтожается, а Page Cache ОС остаётся «горячим» в оперативной памяти сервера, и брокер мгновенно возвращается к пиковой скорости отдачи данных.

4. Что такое «Tombstone» в топике с политикой `cleanup.policy=compact`, и что произойдет, если значение `delete.retention.ms` будет выставлено слишком маленьким?