📨 Раздел 15 · Вопрос #21

Что такое batch в Kafka producer

Вместо того чтобы совершать системный вызов и отправлять TCP-пакет на каждый вызов producer.send(), Kafka Client помещает сериализованные байты во внутренний кольцевой буфер в п...

📨 Kafka 📄 Исходный Markdown

🟢 Junior Level

Что такое Batching в Kafka Producer

Batching (пакетирование) в Apache Kafka Producer — это архитектурный механизм агрегации нескольких записей (ProducerRecord), предназначенных для одной и той же партиции топика, в единый сетевой пакет данных (ProducerBatch) перед их физической передачей брокеру через сокет.

Вместо того чтобы совершать системный вызов и отправлять TCP-пакет на каждый вызов producer.send(), Kafka Client помещает сериализованные байты во внутренний кольцевой буфер в памяти. Фоновый поток ввода-вывода (Sender) извлекает накопленный пакет и передает его лидер-брокеру за один сетевой round-trip (RTT).

Без Batching (RPS = 10 000):
App Thread ──send()──> TCP Packet 1 (IP + TCP + Kafka Header ~100B + Payload 100B) ──> Broker
App Thread ──send()──> TCP Packet 2 ────────────────────────────────────────────────> Broker
App Thread ──send()──> TCP Packet 3 ────────────────────────────────────────────────> Broker
Результат: 10 000 сетевых прерываний, 10 000 дисковых fsync на брокере, гигантский overhead.

С механизмом Batching:
App Thread 1 ──send()─┐
App Thread 2 ──send()─┼─> [RecordAccumulator: ProducerBatch 64KB] ──Sender Thread──> 1 TCP Запрос ──> Broker
App Thread 3 ──send()─┘
Результат: Высокая утилизация сети, блочная запись на диск, сжатие блока целиком.

Зачем нужен Batching

  1. Драматический рост пропускной способности (Throughput): Накладные расходы протоколов IP, TCP и заголовков Kafka Request Protocol амортизируются на тысячи сообщений.
  2. Эффективное потоковое сжатие (Compression): Алгоритмы сжатия (LZ4, ZSTD, Snappy, GZIP) работают на уровне словарей повторяющихся подстрок. Сжатие отдельного JSON-сообщения размером 200 байт практически бесполезно (иногда размер даже растет из-за метаданных). Сжатие пакета из 200 таких сообщений (40 КБ) дает степень компрессии в 3–7 раз, радикально разгружая сеть брокера и память page cache.
  3. Снижение нагрузки на диск брокера: Брокер сбрасывает сетевой буфер прямо в каналы FileChannel операционной системы последовательными блоками, не вызывая фрагментации сегментов лога .log.

Базовые конфигурационные параметры

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker1:9092,kafka-broker2:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

// 1. Порог размера батча в байтах (по умолчанию 16384 = 16 KB)
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 65536); // 64 KB для высокой нагрузки

// 2. Искусственная задержка отправки для накопления батча (по умолчанию 0 ms)
props.put(ProducerConfig.LINGER_MS_CONFIG, 10); // Ждать до 10 миллисекунд

// 3. Общий пул оперативной памяти продюсера (по умолчанию 33554432 = 32 MB)
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864L); // 64 MB

// 4. Алгоритм сжатия батча
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

🟡 Middle Level

Жизненный цикл сообщения и условия отправки пакета

Когда приложение вызывает producer.send(record, callback), выполнение происходит следующим образом:

  1. Выполняются перехватчики (ProducerInterceptor).
  2. Сериализаторы (Serializer) преобразуют ключ и значение в массивы байтов byte[].
  3. Партиционер (Partitioner) вычисляет целевую партицию.
  4. Запись помещается в RecordAccumulator.

Пакет (ProducerBatch) переходит в состояние готовности к отправке (ready) при выполнении любого из следующих условий:

  1. Превышение порога размера (batch.size): Добавление новой записи заполнило внутренний буфер пакета до установленного лимита (например, 64 КБ). Пакет немедленно «запечатывается» и готов к отправке потоком Sender.
  2. Истечение тайм-аута ожидания (linger.ms): Первая запись попала в пустой батч в момент времени $T_0$. Если за время $(T_0 + \text{linger.ms})$ батч не достиг batch.size, он всё равно принудительно помечается готовым к отправке, защищая систему от неограниченного роста задержки (latency).
  3. Исчерпание буферной памяти (buffer.memory): Если пул свободной памяти BufferPool истощен, поток-писатель может форсировать отправку уже накопившихся неполных пакетов для освобождения страниц памяти.
  4. Явный вызов producer.flush(): Приложение блокируется до тех пор, пока абсолютно все находящиеся в аккумуляторе батчи не будут отправлены брокерам и подтверждены ack.
  5. Закрытие клиента producer.close(): Инициируется штатный дренаж всех буферов в пределах close(Duration timeout).

Взаимодействие batch.size и linger.ms

Сценарий batch.size linger.ms Поведение системы Область применения
Ultra Low Latency 16 KB (default) 0 (отправка мгновенно) Сообщения отправляются сразу при готовности сетевого сокета. Батчинг возникает только если сеть временно занята отправкой предыдущего запроса. Финансовый трейдинг, телеком-сигнализация.
High Throughput 64 KB – 128 KB 5 – 20 ms Поток Sender ждет до $N$ миллисекунд, позволяя параллельным потокам приложения наполнить батч до десятков килобайт. Клики, логи, метрики, ETL/CDC-пайплайны.
Misconfiguration 1 1 KB (маленький) 50 ms Батч заполняется мгновенно за 1–2 записи, параметр linger.ms фактически игнорируется, сеть забита микро-запросами. Ошибка проектирования.
Misconfiguration 2 1 MB (огромный) 0 ms Буфер выделен гигантский, но при низкой нагрузке батчи улетают полупустыми из-за linger.ms=0. При всплеске память резко деградирует. Неосознанное копирование параметров.

Поведение при исчерпании буфера: buffer.memory и max.block.ms

Память buffer.memory (по умолчанию 32 МБ) разделяется между всеми батчами для всех партиций, которые обслуживает данный экземпляр продюсера.

Если сеть брокера деградировала, лидер-брокер перестал отвечать или скорость генерации данных приложением превышает скорость сетевого интерфейса:

  1. Свободные сегменты в BufferPool заканчиваются.
  2. Последующие вызовы producer.send() блокируют вызывающий поток приложения.
  3. Время ожидания ограничено параметром max.block.ms (по умолчанию 60 000 мс — 1 минута).
  4. Если за время max.block.ms память не освободилась (например, не пришли ответы от брокера на отправленные батчи), метод send() выбрасывает исключение: org.apache.kafka.common.errors.TimeoutException: Failed to allocate memory within the configured max blocking time.

🔴 Senior Level

Внутреннее устройство RecordAccumulator и BufferPool

На уровне исходного кода Kafka Client (org.apache.kafka.clients.producer.internals) механизм пакетирования реализован на двух фундаментальных структурах: RecordAccumulator и BufferPool.

                KafkaProducer.send() (App Threads)
                           │
                           ▼
                  RecordAccumulator
                           │
          ConcurrentMap<TopicPartition, Deque<ProducerBatch>>
          ┌─────────────────────────────────────────────────┐
          │ "orders-0": [Batch 1 (full)] -> [Batch 2 (open)] │
          │ "orders-1": [Batch 1 (open)]                    │
          │ "users-0":  [Batch 1 (open)]                    │
          └─────────────────────────────────────────────────┘
                           │
                  Запрос байтового буфера
                           │
                           ▼
                      BufferPool
          ┌─────────────────────────────────────────────────┐
          │ totalMemory = 32 MB                             │
          │ poolableSize = batch.size (16 KB)               │
          │ freeSlots: ArrayDeque<ByteBuffer> [16KB, 16KB..]│
          │ nonPooledAvailableMemory: long                  │
          │ waiters: ArrayDeque<Condition>                  │
          └─────────────────────────────────────────────────┘

Анатомия BufferPool и Zero-GC аллокация

  1. Переиспользование буферов: Чтобы предотвратить катастрофическое давление на Garbage Collector (создание и удаление сотен тысяч массивов byte[] в секунду), Kafka использует собственный пул BufferPool.
  2. Если требуемый размер буфера равен в точности batch.size, пул возвращает уже существующий предварительно аллоцированный прямой или JVM-хиповый ByteBuffer из внутреннего списка freeSlots (ArrayDeque<ByteBuffer>).
  3. При завершении отправки батча и получении ответа от брокера вызывается метод deallocate(). Буфер очищается (buffer.clear()) и возвращается обратно в freeSlots. Аллокации памяти в Java Heap не происходит!
  4. Проблема негабаритных записей (Oversized Records): Если размер одного сериализованного сообщения превышает batch.size (например, сообщение 200 КБ при batch.size = 64 KB):
    • BufferPool не может предоставить буфер из пула freeSlots.
    • Пул принудительно аллоцирует нестандартный ByteBuffer через ByteBuffer.allocate(recordSize).
    • После отправки такого батча буфер не возвращается в пул, а выбрасывается сборщику мусора (GC garbage). Если приложение регулярно отправляет записи крупнее batch.size, JVM начинает испытывать сильнейшее давление на Eden/Old Gen память, приводя к частым Stop-the-World паузам.

Потоковая модель: Мультиплексирование в Sender Thread

В продюсере Kafka потоки разделены:

  • Потоки приложения (User Threads): Сериализуют записи и добавляют их в блокирующий двухсвязный список Deque<ProducerBatch> соответствующей TopicPartition. Блокировка кратковременна (на уровне synchronized-секции конкретной очереди деки).
  • Фоновый поток Sender (I/O Daemon Thread):
    • Выполняет итерацию по RecordAccumulator, отбирая все готовые (ready) батчи.
    • Группирует батчи по брокерам-лидерам (NodeId). Если у брокера №1 находятся лидер-партиции topicA-0, topicA-1 и topicB-3, готовые батчи для этих трех партиций объединяются в один сетевой запрос ClientRequest (структура ProduceRequest).
    • Передает запрос во внутренний NetworkClient на базе неблокирующего Java NIO Selector.

Влияние Sticky Partitioner (KIP-480 и KIP-794)

До Kafka 2.4 при отправке сообщений без ключа (key == null) использовался наивный Round-Robin: сообщение 1 шло в партицию 0, сообщение 2 — в партицию 1, сообщение 3 — в партицию 2.

  • Катастрофа для батчинга: Если в топике 50 партиций, а batch.size = 16 KB, продюсеру требовалось сгенерировать 50 × 16 КБ = 800 КБ данных, чтобы отправить полные батчи. В результате сообщения улетали мелкими пакетами по тайм-ауту linger.ms, убивая пропускную способность.

Начиная с Kafka 2.4 (KIP-480) и усовершенствованный в Kafka 3.3 (KIP-794), введен Uniform Sticky Partitioner:

  • Записи без ключа временно «прилипают» к одной конкретной партиции до тех пор, пока батч для этой партиции не заполнится до batch.size или не истечет linger.ms.
  • Как только батч заполнен и отправлен, партиционер переключается на следующую случайную активную партицию.
  • Результат: Задержка уменьшилась на 30–50%, а плотность заполнения батчей выросла до 95%+ без изменения логики клиентского кода.
Round-Robin (Legacy):
Message 1 ──> Partition 0 [msg1 (неполный)] ──таймаут linger.ms──> Отправка 1 записи
Message 2 ──> Partition 1 [msg2 (неполный)] ──таймаут linger.ms──> Отправка 1 записи
Message 3 ──> Partition 2 [msg3 (неполный)] ──таймаут linger.ms──> Отправка 1 записи

Sticky Partitioner (Современный):
Message 1 ──┐
Message 2 ──┼─> Partition 0 [msg1, msg2, msg3, ... msgN] ──batch.size полон──> Отправка 64 КБ
Message 3 ──┘
(Переключение на Partition 1 для следующей пачки)

Гарантии доставки и Idempotent Batching

При включении идемпотентности (enable.idempotence = true, включена по умолчанию начиная с Kafka 3.0):

  • Каждому батчу брокер сопоставляет идентификатор продюсера ProducerId (PID) и монотонно возрастающий порядковый номер SequenceNumber первой записи в батче (baseSequence).
  • Брокер сохраняет в памяти и снэпшотах лога последние 5 последовательностей для каждой партиции (max.in.flight.requests.per.connection <= 5).
  • Если батч завершился сетевой ошибкой таймаута и продюсер повторяет отправку (retries > 0), брокер распознает дубликат пакета по связке (ProducerId, Epoch, baseSequence) и отвергает повторную запись без генерации ошибки для клиента.

4 Tricky Questions

1. Что произойдет, если размер единичного сообщения превышает batch.size, но меньше max.request.size?

Ответ: Сообщение будет успешно отправлено, но с критическими побочными эффектами для производительности:

  1. BufferPool обнаружит, что запрошенный размер буфера превышает фиксированный poolableSize (batch.size).
  2. Пул выделит нестандартный ByteBuffer через оператор аллокации ByteBuffer.allocate(recordSize) из непулируемой памяти (nonPooledAvailableMemory).
  3. Для этого единственного сообщения будет сформирован отдельный ProducerBatch, размер которого равен размеру сообщения. Этот батч сразу закрывается для добавления новых записей и помечается готовым к отправке.
  4. После отправки буфер не может быть возвращен в пул переиспользуемых сегментов freeSlots и выбрасывается в Garbage Collector.
  5. Последствия: Если такие сообщения идут постоянным потоком, пул деградирует в обычный генератор мусора в куче, провоцируя частые GC-паузы и фрагментацию памяти. Если сообщение превышает max.request.size (по умолчанию 1 МБ), продюсер сразу выбросит RecordTooLargeException без попытки аллокации.

2. Приложение вызывает producer.send() с linger.ms = 100, однако батчи отправляются каждые 0.5 мс, не заполняясь до batch.size. В чем причина?

Ответ: Существует несколько взаимосвязанных причин:

  1. Медленная генерация или синхронная отправка: Если вызывающий код делает producer.send(record).get(), поток блокируется до получения ответа брокера. В этом случае в аккумуляторе физически не может накопиться более одного сообщения.
  2. Параметр max.in.flight.requests.per.connection = 1 при постоянной отправке: Как только предыдущий запрос завершился, поток Sender просыпается по событию epoll/kqueue селектора. Если в очереди партиции уже лежит хотя бы одно сообщение, Sender может захватить его для отправки в освободившееся соединение, не выжидая linger.ms.
  3. Истощение buffer.memory: Когда общий объем свободной памяти падает до нуля, продюсер принудительно пробуждает поток Sender для дренажа доступных неполных батчей, чтобы разблокировать ожидающие потоки приложения.
  4. Вызов flush() из другого потока: Любой параллельный вызов producer.flush() мгновенно форсирует отправку всех готовых и неготовых батчей.

3. Как взаимодействуют параметры delivery.timeout.ms, request.timeout.ms, linger.ms и повторные попытки (retries)?

Ответ: Начиная с Kafka 2.0 введен глобальный параметр контроля времени жизни сообщения — delivery.timeout.ms (по умолчанию 120 000 мс — 2 минуты). Математическое правило валидации клиента: \(\text{delivery.timeout.ms} \ge \text{linger.ms} + \text{request.timeout.ms}\)

  • Жизненный цикл таймера: Таймер батча стартует не в момент физической отправки в сокет, а в момент помещения первой записи в батч внутри RecordAccumulator.
  • Время батча в очереди = время ожидания накопления (linger.ms или заполнение) + время ожидания в очереди сокета до отправки.
  • Если батч был отправлен, но брокер ответил ошибкой (например, NOT_ENOUGH_REPLICAS), батч возвращается в голову очереди деки партиции.
  • Батч будет повторять попытки (retries), пока суммарное время с момента создания батча не превысит delivery.timeout.ms. По истечении этого времени callback завершается с TimeoutException, независимо от того, сколько попыток retries было выполнено.

4. Почему сжатие (compression.type) применяется на уровне батча, а не всего сетевого запроса (ClientRequest)?

Ответ: Это ключевое архитектурное решение Kafka для обеспечения принципа Zero-Copy на брокере:

  1. Сетевой запрос продюсера (ClientRequest) может содержать десятки батчей для разных топиков и партиций, лидером которых является данный брокер.
  2. Если бы сжатие происходило на уровне всего сетевого запроса, брокер был бы обязан полностью декомпрессировать сетевой пакет в оперативную память CPU, рассортировать записи по партициям и сегментам лога на диске, а затем снова компрессировать каждую партицию перед записью на диск. Это вызвало бы колоссальную утилизацию CPU брокера.
  3. Поскольку сжатие выполняется продюсером индивидуально для каждого ProducerBatch:
    • Брокер выполняет валидацию заголовка батча.
    • Сжатые байты ProducerBatch с помощью системного вызова сбрасываются в файл лога партиции (.log) напрямую без распаковки.
    • При чтении консьюмером брокер читает файл лога через sendfile(2) (Zero-Copy) и пересылает сжатые байты по сети консьюмеру.
    • Распаковку выполняет только конечный потребитель (KafkaConsumer).

🎯 Шпаргалка для интервью

30-секундный ответ (Elevator Pitch)

«Batch в Kafka Producer — это механизм накопления сообщений для одной партиции в буфере памяти (RecordAccumulator) перед отправкой единым блоком брокеру. Он управляется двумя ключевыми регуляторами: batch.size (лимит объема в байтах, по умолчанию 16 КБ) и linger.ms (максимальное время искусственной задержки накопления, по умолчанию 0 мс). Пакет отправляется, как только наступит первое из событий: заполнен batch.size или истек linger.ms. Batching радикально увеличивает пропускную способность за счет снижения сетевых накладных расходов, обеспечивает эффективное блочное сжатие (LZ4/ZSTD) и снижает дисковую фрагментацию на брокерах благодаря Zero-Copy транзиту сжатых блоков».

Ключевые метрики для Production-мониторинга

  • batch-size-avg / batch-size-max: средний и максимальный размер отправленных пакетов в байтах. Если средний размер близок к размеру одного сообщения — батчинг не работает (нужно увеличить linger.ms).
  • record-queue-time-avg: среднее время нахождения записи в RecordAccumulator.
  • bufferpool-wait-time-ns / buffer-exhausted-rate: частота блокировок потоков из-за нехватки буферной памяти продюсера (buffer.memory). Должна быть строго равна 0.
  • records-per-request-avg: среднее число сообщений в одном сетевом запросе к брокеру.

Красные флаги (Чего говорить НЕЛЬЗЯ)

  • ❌ «Батч объединяет сообщения со всего продюсера в один пакет» — Нет, батч формируется строго в разрезе конкретной партиции (TopicPartition). Разные партиции имеют разные очереди батчей.
  • ❌ «Если поставить linger.ms = 0, батчинга не будет никогда» — Неверно. Если сеть перегружена или предыдущий сетевой запрос еще находится в сокете (in-flight), последующие сообщения успевают сбатчеваться даже при linger.ms = 0.
  • ❌ «При buffer.memory = 32MB продюсер всегда может отправить сообщение размером 33MB» — Нет, максимальный размер одного сообщения ограничен max.request.size (1 МБ) и лимитом брокера message.max.bytes. Сообщение больше buffer.memory немедленно завершится ошибкой.
  • ❌ «Брокер распаковывает батч, проверяет каждое сообщение и запаковывает обратно» — Грубая ошибка. Брокер записывает сжатый батч на диск “as is”, распаковка происходит только на консьюмере.

Связанные темы