📨 Раздел 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.

Связанные вопросы