Що таке batch в Kafka producer

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

🟢 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

Драматичне зростання пропускної здатності (Throughput): Накладні витрати протоколів IP, TCP і заголовків Kafka Request Protocol амортизуються на тисячі повідомлень.
Ефективне потокове стиснення (Compression): Алгоритми стиснення (LZ4, ZSTD, Snappy, GZIP) працюють на рівні словників повторюваних підрядків. Стиснення окремого JSON-повідомлення розміром 200 байт практично марне (іноді розмір навіть зростає через метадані). Стиснення пакета з 200 таких повідомлень (40 КБ) дає коефіцієнт компресії в 3–7 разів, радикально розвантажуючи мережу брокера та пам’ять page cache.
Зниження навантаження на диск брокера: Брокер скидає мережевий буфер безпосередньо в канали 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), виконання відбувається наступним чином:

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

Пакет (ProducerBatch) переходить у стан готовності до відправки (ready) при виконанні будь-якої з таких умов:

Перевищення порогу розміру (batch.size): Додавання нового запису заповнило внутрішній буфер пакета до встановленого ліміту (наприклад, 64 КБ). Пакет негайно «запечатується» і готовий до відправки потоком Sender.
Спливання таймауту очікування (linger.ms): Перший запис потрапив у порожній батч у момент часу $T_0$. Якщо за час $(T_0 + \text{linger.ms})$ батч не досяг batch.size, він усе одно примусово позначається готовим до відправки, захищаючи систему від необмеженого зростання затримки (latency).
Вичерпання буферної пам’яті (buffer.memory): Якщо пул вільної пам’яті BufferPool вичерпано, потік-письменник може форсувати відправку вже накопичених неповних пакетів для звільнення сторінок пам’яті.
Явний виклик producer.flush(): Застосунок блокується доти, доки абсолютно всі батчі, що знаходяться в акумуляторі, не будуть відправлені брокерам і підтверджені ack.
Закриття клієнта 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-пайплайни.
Помилка конфігурації 1	`1 KB` (замалий)	`50 ms`	Батч заповнюється миттєво за 1–2 записи, параметр `linger.ms` фактично ігнорується, мережа забита мікро-запитами.	Помилка проєктування.
Помилка конфігурації 2	`1 MB` (величезний)	`0 ms`	Буфер виділено гігантський, але при низькому навантаженні батчі відлітають напівпорожніми через `linger.ms=0`. При сплеску пам’ять деградує.	Неусвідомлене копіювання параметрів.

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

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

Якщо мережа брокера деградувала, лідер-брокер перестав відповідати або швидкість генерації даних застосунком перевищує пропускну здатність мережевого інтерфейсу:

Вільні сегменти в BufferPool закінчуються.
Подальші виклики producer.send() блокують викликаючий потік застосунку.
Час очікування обмежений параметром max.block.ms (за замовчуванням 60 000 мс — 1 хвилина).
Якщо за час 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 алокація

Перевикористання буферів: Щоб запобігти катастрофічному тиску на Garbage Collector (створення та видалення сотень тисяч масивів byte[] за секунду), Kafka використовує власний пул BufferPool.
Якщо необхідний розмір буфера дорівнює в точності batch.size, пул повертає вже існуючий попередньо виділений прямий або JVM-хіповий ByteBuffer із внутрішнього списку freeSlots (ArrayDeque<ByteBuffer>).
При завершенні відправки батча та отриманні відповіді від брокера викликається метод deallocate(). Буфер очищається (buffer.clear()) і повертається назад у freeSlots. Алокації пам’яті в Java Heap не відбувається!
Проблема негабаритних записів (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`?

Відповідь: Повідомлення буде успішно відправлено, але з критичними побічними ефектами для продуктивності:

BufferPool виявить, що запитаний розмір буфера перевищує фіксований poolableSize (batch.size).
Пул виділить нестандартний ByteBuffer через оператор виділення ByteBuffer.allocate(recordSize) із непульованої пам’яті (nonPooledAvailableMemory).
Для цього єдиного повідомлення буде сформовано окремий ProducerBatch, розмір якого дорівнює розміру повідомлення. Цей батч одразу закривається для додавання нових записів і позначається готовим до відправки.
Після відправки буфер не може бути повернений до пулу перевикористовуваних сегментів freeSlots і викидається збирачу сміття (Garbage Collector).
Наслідки: Якщо такі повідомлення йдуть постійним потоком, пул деградує у звичайний генератор сміття в купі, провокуючи часті GC-паузи та фрагментацію пам’яті. Якщо повідомлення перевищує max.request.size (за замовчуванням 1 МБ), продюсер одразу викине RecordTooLargeException без спроби алокації.

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

Відповідь: Існує декілька взаємопов’язаних причин:

Синхронна відправка: Якщо викликаючий код робить producer.send(record).get(), потік блокується до отримання відповіді брокера. У цьому разі в акумуляторі фізично не може накопичитися більше одного повідомлення.
Параметр max.in.flight.requests.per.connection = 1 при постійній відправці: Щойно попередній запит завершився, потік Sender прокидається за подією epoll/kqueue селектора. Якщо в черзі партиції вже лежить хоча б одне повідомлення, Sender може негайно захопити його для відправки у звільнене з’єднання, не вичікуючи linger.ms.
Виснаження buffer.memory: Коли загальний обсяг вільної пам’яті падає до нуля, продюсер примусово пробуджує потік Sender для дренажу доступних неповних батчів, щоб розблокувати очікуючі потоки застосунку.
Виклик 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 на брокері:

Мережевий запит продюсера (ClientRequest) може містити десятки батчів для різних топіків і партицій, лідером яких є даний брокер.
Якби стиснення відбувалося на рівні всього мережевого запиту, брокер був би зобов’язаний повністю декомпресувати мережевий пакет в оперативну пам’ять CPU, розсортувати записи по партиціях і сегментах лога на диску, а потім знову компресувати кожну партицію перед записом на диск. Це викликало б колосальне навантаження на CPU брокера.
Оскільки стиснення виконується продюсером індивідуально для кожного 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”, розпакування відбувається лише на консьюмері.