📨 Розділ 15 · Питання #3

Як дані розподіляються по партиціях

Розподіл повідомлень по партиціях топіка виконується на стороні клієнта-продюсера компонентом Partitioner за двома базовими сценаріями:

📨 Kafka 📄 Вихідний Markdown

🟢 Junior Level

30-секундна відповідь

Розподіл повідомлень по партиціях топіка виконується на стороні клієнта-продюсера компонентом Partitioner за двома базовими сценаріями:

  1. Повідомлення з ключем (key != null): брокер використовує детерміноване хешування. Хеш-код ключа (алгоритм MurmurHash2) ділиться за модулем на загальну кількість партицій. Усі повідомлення з однаковим ключем (наприклад, orderId = "ord-100") завжди потрапляють суворо в одну й ту саму партицію, гарантуючи порядок обробки FIFO.
  2. Повідомлення БЕЗ ключа (key == null): починаючи з Kafka 2.4+, використовується Sticky Partitioning (липке партиціювання). Продюсер пакує повідомлення пачками (batches) в одну обрану партицію до її наповнення, після чого перемикається на наступну. Це максимізує стиснення та пропускну здатність мережі.

Наочна аналогія з життя

Розподіл повідомлень — це сортування посилок на поштовому складі:

  • З ключем: на кожній коробці написано поштовий індекс регіону (ключ). Сортувальник завжди відправляє коробки з індексом «101» суворо у фургон №1. Скільки б коробок із цим індексом не надійшло, вони всі поїдуть в одній машині та прибудуть у порядку відправлення.
  • Без ключа: коробки без індексу (загальний вантаж). Сортувальник не розкидає по одній коробці в різні фургони (старий Round-Robin), а повністю завантажує фургон №1 до стелі (Sticky Partitioning), зачиняє двері й лише потім переходить до фургона №2.

Базовий приклад відправлення в Java

// 1. Відправлення З КЛЮЧЕМ (детермінована маршрутизація)
ProducerRecord<String, String> keyedRecord = 
    new ProducerRecord<>("orders", "customer-42", "{\"item\":\"phone\"}");
producer.send(keyedRecord); // Завжди одна й та сама партиція для customer-42

// 2. Відправлення БЕЗ КЛЮЧА (Sticky Partitioning)
ProducerRecord<String, String> unkeyedRecord = 
    new ProducerRecord<>("telemetry", null, "{\"temp\": 23.5}");
producer.send(unkeyedRecord); // Пакується в поточний активний батч

🟡 Middle Level

Алгоритм DefaultPartitioner та математична пастка

Під капотом стандартний DefaultPartitioner обчислює цільову партицію за формулою: \(\text{partition} = \text{Utils.toPositive}(\text{Utils.murmur2}(\text{keyBytes})) \pmod{\text{numPartitions}}\)

Чому не Math.abs(key.hashCode()) % numPartitions?

Це одна з улюблених тем технічних співбесід з Java:

// ❌ ГРУБА ПОМИЛКА в самописних партиціонерах:
int partition = Math.abs(key.hashCode()) % numPartitions;

У Java діапазон int від $-2^{31}$ до $2^{31}-1$. Для числа Integer.MIN_VALUE ($-2147483648$) у доповняльному двійковому коді немає додатного аналога! Виклик Math.abs(Integer.MIN_VALUE) повертає від’ємне число: Integer.MIN_VALUE! Операція % numPartitions повертає від’ємний індекс партиції, що призводить до падіння застосунку з ArrayIndexOutOfBoundsException! Kafka вирішує цю проблему побітовою маскою в службовому методі:

public static int toPositive(int number) {
    return number & 0x7fffffff; // Обнуляє знаковий біт, роблячи число суворо >= 0
}

Еволюція маршрутизації без ключа: Round-Robin vs Sticky Partitioning

[ Старий Round-Robin (до Kafka 2.4) ]
Запит 1 ──► Партиція 0 (батч 100 байтів) ──► Відправлення в мережу
Запит 2 ──► Партиція 1 (батч 100 байтів) ──► Відправлення в мережу
Запит 3 ──► Партиція 2 (батч 100 байтів) ──► Відправлення в мережу
Підсумок: мережа забита мікро-пакетами, стиснення не працює, високий CPU оверхед.

[ Sticky Partitioning (Kafka 2.4+) ]
Запити 1, 2, 3... 50 ──► Усі пишуться в Партицію 0
(Батч досяг batch.size = 16KB або сплив linger.ms) ──► Одне щільне мережеве відправлення
Запит 51... ──► Перемикання на Партицію 1
Підсумок: екстремальна пропускна здатність, відмінне стиснення LZ4/ZSTD.

Написання Custom Partitioner

Якщо бізнес-вимоги потребують спеціальної маршрутизації (наприклад, VIP-клієнти завжди йдуть у партицію 0, а решта розподіляється по партиціях 1..N):

public class VipCustomerPartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();

        if (key instanceof String stringKey && stringKey.startsWith("VIP_")) {
            return 0; // Виділена партиція для преміальних клієнтів
        }

        // Решта партицій (1 .. numPartitions - 1)
        int nonVipPartitions = numPartitions - 1;
        int hash = Utils.toPositive(Utils.murmur2(keyBytes));
        return 1 + (hash % nonVipPartitions);
    }

    @Override
    public void configure(Map<String, ?> configs) {}

    @Override
    public void close() {}
}

🔴 Senior Level

Проблема «Гарячого ключа» (Hot Key Skew) і техніки боротьби

У великих системах розподіл ключів підпорядковується закону Парето: 1% клієнтів генерують 90% навантаження.

  • Якщо ключ — це merchantId, а великий ритейлер генерує 50 000 повідомлень на секунду, його партиція захлинеться, а консьюмер відстане (Consumer Lag).

Архітектурні шаблони усунення Skew:

  1. Compound Routing Key (Складений бізнес-ключ): Якщо всередині одного клієнта нам потрібен порядок лише в межах конкретного замовлення, ключем призначається зв’язка: \(\text{key} = \text{merchantId} + \text{"\#"} + \text{orderId}\) Замовлення одного мерчанта рівномірно розподіляються по всіх партиціях, зберігаючи при цьому суворий порядок подій за кожним конкретним замовленням!
  2. Key Salting з обмеженням за модулем (Ограничене соління): 
    // Розщеплюємо супер-важкий ключ рівно на 4 партиції:
int salt = ThreadLocalRandom.current().nextInt(4);
String routedKey = "SUPER_MERCHANT_" + salt;

KIP-794: Архітектурні зміни в Kafka 3.3+

Починаючи з Apache Kafka 3.3.0, інтерфейс Partitioner і конфігурація partitioner.class були визнані deprecated:

  • У сучасних версіях логіка вибору партицій вбудована безпосередньо в KafkaProducer через оптимізований клас BuiltInPartitioner.
  • Це дозволило усунути зайві алокації об’єктів у пам’яті JVM на кожне повідомлення, що відправляється, і реалізувати адаптивний алгоритм Adaptive Partitioning: продюсер динамічно відстежує latency брокерів і починає віддавати перевагу тим партиціям, лідери яких відповідають швидше, уникаючи відправлення повідомлень у перевантажені вузли.

Вплив linger.ms та batch.size на Sticky Partitioning

  • batch.size (за замовчуванням 16 КБ): максимальний розмір батча в байтах.
  • linger.ms (за замовчуванням 0 мс): час очікування накопичення батча перед примусовим відправленням у сокет.

Небезпечний граничний випадок (Edge Case): Якщо в продюсері виставлено параметри linger.ms = 0 та batch.size = 16384, але потік подій низький (1 повідомлення раз на 50 мс):

  • Кожне повідомлення миттєво відправляється в мережу окремим мережевим запитом.
  • Sticky Partitioner після відправлення незаповненого батча виконує ротацію партиції.
  • У результаті Sticky Partitioning фактично вироджується в неефективний Round-Robin, генеруючи величезну кількість напівпорожніх мережевих пакетів TCP. Для високонавантажених продюсерів обов’язковим є встановлення linger.ms = 10..20 мс.

4 Tricky Questions

1. Чому популярний код Math.abs(key.hashCode()) % partitions.size() періодично призводив до падіння сервісів у Production з помилкою ArrayIndexOutOfBoundsException?

Відповідь:

  • У Java метод Math.abs(int a) для від’ємних чисел виконує операцію -a.
  • Мінімальне від’ємне 32-бітне число Integer.MIN_VALUE дорівнює $-2147483648$.
  • Максимальне додатне число дорівнює $2147483647$.
  • Оскільки число $2147483648$ не вміщується в додатний діапазон int, при переповненні знаковий біт залишається рівним 1, і метод Math.abs(Integer.MIN_VALUE) повертає $-2147483648$!
  • У результаті операція % partitions.size() дає від’ємний залишок (наприклад, -2), і при спробі звернутися до partitions.get(-2) сервіс викидає ArrayIndexOutOfBoundsException.
  • Рішення: використовувати побітову маску key.hashCode() & 0x7fffffff або утиліту Kafka Utils.toPositive().

2. Чому Kafka використовує алгоритм MurmurHash2 для обчислення хешу ключа замість стандартного String.hashCode() з Java SDK?

Відповідь: З двох фундаментальних причин:

  1. Мовна нейтральність (Polyglot Clients): Kafka — це розподілена платформа. Продюсери можуть бути написані на Go, C++, Python, Rust або C#. Стандартний метод Java String.hashCode() специфічний для JVM. Алгоритм MurmurHash2 є загальноприйнятим індустріальним стандартом з єдиним байтовим алгоритмом у всіх мовах програмування. Завдяки цьому Go-продюсер і Java-продюсер відправлять повідомлення з ключем "order-1" в одну й ту саму партицію.
  2. Рівномірність розподілу (Avalanche Effect): String.hashCode() страждає від високої частоти колізій на коротких або схожих рядках. MurmurHash2 має чудовий лавинний ефект (зміна 1 біта в ключі змінює 50% бітів у хеші), забезпечуючи ідеальний розподіл по партиціях без перекосів.

3. Як зміниться маршрутизація повідомлень, якщо відправляти записи з ключем типу byte[] у порівнянні з ключем типу String?

Відповідь:

  • Якщо передається String, стандартний серіалізатор StringSerializer перетворює рядок у масив байтів у кодуванні UTF-8, і MurmurHash2 обчислюється від цих байтів.
  • Якщо передається кастомний Java-об’єкт без кастомного серіалізатора, або якщо в Java перевизначено hashCode(), Kafka Partitioner ніколи не викликає Java-метод .hashCode() в об’єкта!
  • Partitioner Kafka завжди працює виключно із серіалізованим бінарним масивом byte[].
  • Увага: якщо серіалізатор серіалізує Java-об’єкти недетерміновано (наприклад, використовуючи стандартну серіалізацію Java з різними метаданими полів або JSON із плаваючим порядком полів), однакові за змістом об’єкти отримають різні байтові масиви, різні хеші та розлетяться по різних партиціях!

4. Що станеться зі Sticky Partitioner, якщо цільовий брокер, на якому знаходиться лідер поточної партиції, раптово починає відповідати із затримками (I/O Bottleneck)?

Відповідь:

  • У версіях Kafka до 3.2 продюсер продовжував уперто надсилати повідомлення в поточну «залиплу» партицію, доки не заповниться батч, що призводило до зростання черг у RecordAccumulator і вичерпання пам’яті buffer.memory на стороні продюсера.
  • Починаючи з Kafka 3.3+ (у межах KIP-794), вбудований партиціонер відстежує час очікування в черзі та готовність брокерів. Якщо партиція блокується або брокер не відповідає протягом таймауту, партиціонер примусово перемикає липкий батч (Unstick) на іншу доступну партицію здорового брокера, захищаючи продюсер від каскадної відмови за таймаутами відправлення.

🎯 Шпаргалка для інтерв’ю

Головне про розподіл даних

  • З ключем: Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions $\to$ суворий детермінізм і FIFO.
  • Без ключа: Sticky Partitioning (з Kafka 2.4+) $\to$ батчинг в одну партицію до заповнення, потім ротація.
  • Математика Java: Math.abs(Integer.MIN_VALUE) < 0! У партиціонерах використовувати hash & 0x7fffffff.
  • MurmurHash2: міжмовний алгоритм, що гарантує ідентичну маршрутизацію з Java, Go, Python.
  • Hot Key Skew: перекіс навантаження від великих клієнтів; лікується солінням ключів (Salting) або складеними ключами.
  • Зміна кількості партицій: ламає відповідність key -> partition для всіх старих ключів!

Стратегії партиціювання

| Метод | Коли застосовується | Порядок повідомлень | Пропускна здатність | | :— | :— | :— | :— | | Keyed (MurmurHash2) | key != null | ✅ Суворий FIFO в межах ключа | Залежить від рівномірності ключів | | Sticky Partitioning | key == null | ❌ Глобального порядку немає | 🚀 Максимальна (великі стислі батчі) | | Round-Robin | Застарів (до 2.4) | ❌ Порядку немає | 🔻 Низька (дрібні неефективні пакети) | | Custom Partitioner | Специфічні правила | Регулюється кодом | Залежить від реалізації |

Червоні прапорці на співбесіді (Чого говорити не можна)

  • ❌ «Якщо ключа немає, Kafka надсилає по одному повідомленню по черзі в кожну партицію (Round-Robin)» — застаріло з 2019 року (Kafka 2.4); за замовчуванням працює Sticky Partitioning.
  • ❌ «Kafka використовує стандартний метод key.hashCode() для обчислення партиції» — ні, використовується MurmurHash2 від серіалізованого масиву байтів.
  • ❌ «Щоб полагодити від’ємний хеш у Java, я просто викликаю Math.abs()» — фатально для Integer.MIN_VALUE.

Пов’язані запитання