Согласно документации Кафки:
Производитель отвечает за выбор того, какое сообщение назначить тому или иному разделу в теме.
Как я могу отправлять сообщения в выбранный раздел, используя kafka-console-producer.sh?
Я хотел бы указать какой-то "идентификатор раздела" при отправке сообщения.
Обновление: этот ответ был правильным в 2014 году, но более современные версии Kafka могут создавать пары ключ/значение через производителя консоли. Смотрите ответы ниже
kafka-console-producer.sh не поддерживает создание сообщений для определенного раздела из коробки.
Однако должно быть довольно просто обновить скрипт, чтобы передать дополнительный параметр для идентификатора раздела, а затем обработать его в пользовательском разделителе, как описано в посте @Chiron в модифицированной версии класса kafka.tools.ConsoleProducer.
Взгляните на исходный код по адресу:
https://apache.googlesource.com/kafka/+/refs/heads/trunk/bin/kafka-console-producer.sh https://apache.googlesource.com/kafka/+/refs/heads/trunk/core/src/main/scala/kafka/tools/ConsoleProducer.scala
В настоящее время ConsoleProducer, похоже, поддерживает запись сообщений с ключами в тему. Kafka будет использовать хэш ключа для распространения сообщения на разделы, по крайней мере, с поведением по умолчанию.
В настоящее время разделителем по умолчанию является \t, поэтому ввод key[\t]message будет распространять его среди разделов:
key1 a-message
Сепаратор можно изменить, предоставив конфигурацию key.separator, например:
kafka-console-producer --broker-list localhost:9092,localhost:9093 \
--topic mytopic --property key.separator=,
Отправить такие сообщения:
key2,another-message
Я тестировал это со стандартной вкладкой и пользовательским seperator успешно. Сообщения были распределены между двумя отдельными разделами.
В соответствии с текущим состоянием вещей (Kafka >= 0.10.0.1) kafka-console-producer.sh script и базовый класс java класса ConsoleProducer поддерживают отправку данных с помощью ключа, но такая поддержка по умолчанию отключена и должен быть включен из CLI.
А именно, вам нужно установить свойство parse.key. Кроме того, если вы хотите использовать что-то отличное от символа табуляции, используйте key.separator, как указано в ответе Седрика.
В конце концов, командной строки будет:
kafka-console.producer.sh --broker-list kafka:9092,kafka2:9092 \
--topic $TOPIC --property parse.key=true --property key.separator=|
Вот ваша отправная точка: partitioner.class в вашем экземпляре Properties. В Kafka реализация по умолчанию kafka.producer.DefaultPartitioner.
Цель этой настройки:
Класс разделителя для разделения сообщений между подтемами. Разделитель по умолчанию основан на хэше ключа.
Это означает, что если вы хотите изменить поведение стандартного разделителя по умолчанию, вам необходимо создать собственную реализацию интерфейса kafka.producer.Partitioner.
Я бы посоветовал быть очень осторожным при создании вашей собственной стратегии и действительно, много тестировать и контролировать свои темы и их разделы.