Как заставить сообщения одного клиента попасть в одну партицию в Kafka

Ответ

В Kafka гарантия попадания сообщений одного клиента в одну партицию достигается использованием ключей (keys) при отправке сообщений. Это критически важно для обеспечения порядка доставки и обработки сообщений.

Механизм работы

Когда вы отправляете сообщение с ключом, Kafka использует функцию хеширования для определения целевой партиции:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

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

// Сообщения с одинаковым ключом всегда попадут в одну партицию
String clientId = "client-123";
ProducerRecord<String, String> record = new ProducerRecord<>(
    "my-topic",
    clientId,  // ключ — определяет партицию
    "message content" // значение
);

producer.send(record);
producer.close();

Как это работает

1. Хеширование ключа: Kafka применяет функцию хеширования к ключу (по умолчанию — муrmur2)
2. Выбор партиции: partition = hash(key) % number_of_partitions
3. Гарантия: Все сообщения с одинаковым ключом попадут в одну и ту же партицию

Примеры использования

Пример 1: Идентификатор пользователя как ключ

String userId = "user-456";
ProducerRecord<String, String> record = new ProducerRecord<>(
    "user-events",
    userId,  // гарантирует, что все события пользователя в одной партиции
    "{\"{event\":\"login\"}"
);
producer.send(record);

Пример 2: Составной ключ

String compositeKey = "user:456:session:789";
ProducerRecord<String, String> record = new ProducerRecord<>(
    "events",
    compositeKey,
    eventData
);
producer.send(record);

Важные детали

• Null ключ: Если отправить сообщение с null ключом, Kafka выберет партицию случайно или использует round-robin
• Изменение количества партиций: При увеличении числа партиций старые сообщения остаются в старых партициях, но новые сообщения могут распределиться по-другому
• Пользовательский партиционер: Можно реализовать Partitioner для кастомной логики:

public class CustomPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes,
                         Cluster cluster) {
        if (key == null) return 0;
        // Кастомная логика определения партиции
        return Math.abs(key.hashCode()) % cluster.partitionsForTopic(topic).size();
    }
}

props.put("partitioner.class", "com.example.CustomPartitioner");

Гарантии при использовании ключей

• Порядок в партиции: Сообщения с одним ключом обрабатываются в порядке отправки
• Масштабируемость: Каждый consumer в группе обрабатывает определённые партиции
• Избежание race conditions: Для обработки данных одного клиента одного потребителем

Можно также настроить acks=all для гарантии доставки на все реплики перед ответом клиенту.