How to produce messages to selected partition using kafka-console-producer?

18
votes

According to the Kafka documentation:

The producer is responsible for choosing which message to assign to which partition within the topic.

How can I send messages to a selected partition using kafka-console-producer.sh?

I would like to specify some sort of 'partition id' at message sending.

5
apache-kafka

5 Answers

19
votes

By now, the ConsoleProducer seems to support writing keyed messages to the topic. Kafka will use the hash of the key to distribute the message into partitions, at least with the default behaviour.

Currently, the default separator is \t, so entering key[\t]message will distribute it amongst partitions:

key1    a-message

The separator can be changed by providing the key.separator configuration, for example:

kafka-console-producer --broker-list localhost:9092,localhost:9093 \
  --topic mytopic --property key.separator=,

Send messages like this:

key2,another-message

I have tested this with the default tab and a custom seperator successfuly. The messages were distributed to two separate partitions.

13
votes

According to the current state of things (Kafka>=0.10.0.1), the kafka-console-producer.sh script and the underlying ConsoleProducer java class support sending data with a key, but such support is disabled by default and has to be enabled from the CLI.

Namely, you need to set the property parse.key. Also, if you want to use something different than a tab character, use key.separator as specified in Cedric's answer.

In the end, the command line would be:

kafka-console.producer.sh --broker-list kafka:9092,kafka2:9092 \
    --topic $TOPIC --property parse.key=true --property key.separator=|
6
votes

Here is your starting point:
partitioner.class setting in your Properties instance. In Kafka, the default implementation is kafka.producer.DefaultPartitioner.

The goal of that setting is:

The partitioner class for partitioning messages amongst sub-topics. The default partitioner is based on the hash of the key.

This means that if you want to change the behaviour of the default partitioner , then you need to create your own implementation of kafka.producer.Partitioner interface.

I would suggest to be really careful when creating your own strategy and really, test it a lot and monitor your topics and their partitions.

2
votes

Update: This answer was correct on 2014, but more current versions of Kafka can produce key/value pairs via the console producer. See the below answers

kafka-console-producer.sh doesn't support producing messages to a particular partition out of the box.

However it should be pretty straightforward to update the script to pass an extra parameter for partition Id and then handle it in a custom partitioner as described in the post by @Chiron in a modified version of kafka.tools.ConsoleProducer class.

Take a look at the source code at:

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

1
votes
C:\arunsingh\demo\kafka_2.13-2.4.0\bin\windows>kafka-console-producer.bat --broker-list 127.0.0.1:9094 --topic arun_topic --property parse.key=true --property key.separator=, --producer-property acks=all
>myKey1, Message with key
>myKey2, Message with key 2
>