Consume and produce message in particular Kafka partition?

Question

For reading all partitions in topic:

~bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic myTopic --from-beginning

1. How can I consume particular partition of the topic? (for instance with partition key 13)
2. And how produce message in partition with particular partition key? Is it possible?

Answer 1

You can't using console consumer and producer. But you can using higher level clients (in any language that works for you).

1. You may use for example assign method to manually assign a specific topic-partition to consume (https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java#L906)
2. You may use a custom Partitioner to override the partitioning logic where you will decide manually how to partition your messages (https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java#L206-L208)

Answer 2

With the many clients that are available you can specify the partition number just like serejja has stated.

Also look into https://github.com/cakesolutions/scala-kafka-client which uses actors and provides multiple modes for manual partitions and offsets.

If you want to do the same on the terminal, I suggest using kafkacat. (https://github.com/edenhill/kafkacat) My personal choice during development.

You can do things like

kafkacat -b localhost:9092 -f 'Topic %t[%p], offset::: %o, data: %s key: %k\n' -t testtopic

And for a specific partition, you just need to use -p flag.

Answer 3

Console producer and consumer do not provide this flexibility. You could achieve this through Kafka APIs.

1. You could manually assign partition to consumer using assign() operation KafkaConsumer/Assign. This will disable group rebalancing. Please use this very carefully.

2. You could specify partition detail in KafkaProducer message. If not specified, it stores as per Partitioner policy.

Answer 4

How can I consume particular partition of the topic? (for instance with partition key 13)

There is a flag called --partition in kafka-console-consumer

--partition <Integer: partition>         The partition to consume from.         
                                           Consumption starts from the end of   
                                           the partition unless '--offset' is   
                                           specified.                           

The command is as follows:

bin/kafka-console-consumer --bootstrap-server localhost:9092 --topic test --partition 0 --from-beginning