# Logstash-output Kafka: message load balancing

**URL:** <https://discuss.elastic.co/t/logstash-output-kafka-message-load-balancing/37804>\
**Category:** Logstash\
**Created:** [December 22, 2015, 10:54pm UTC](https://discuss.elastic.co/t/logstash-output-kafka-message-load-balancing/37804 "2015-12-22T22:54:14Z")\
**Posts on this page:** 5\
**Page:** 1

<div class="post-metadata">

**Author:** ![tguan](https://avatars.discourse-cdn.com/v4/letter/t/958977/32.png) [@tguan](https://discuss.elastic.co/u/tguan)\
**Post date:** [December 22, 2015, 10:54pm UTC](https://discuss.elastic.co/t/logstash-output-kafka-message-load-balancing/37804/1 "2015-12-22T22:54:14Z")

</div>

Is there any good example for specifying the message\_key for logstash Kafka output to balance the messages to be stored between different partitions on the kafka server side? I have 3 partitions in the topic, and running 3 logstash instances. Hope to see each Logstash instance would read messages off different partition. Looks like all the messages are coming to the same partition.

I am using timestamp as the message key from client side. The order of messages is not critical.

Thanks

---

<div class="post-metadata">

**Author:** ![warkolm](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/warkolm/32/39224_2.png) [@warkolm](https://discuss.elastic.co/u/warkolm)\
**Post date:** [December 23, 2015, 10:01pm UTC](https://discuss.elastic.co/t/logstash-output-kafka-message-load-balancing/37804/2 "2015-12-23T22:01:03Z")

</div>

What version of things are you running and what does your config look like?

---

<div class="post-metadata">

**Author:** ![tguan](https://avatars.discourse-cdn.com/v4/letter/t/958977/32.png) [@tguan](https://discuss.elastic.co/u/tguan)\
**Post date:** [December 28, 2015, 11:18pm UTC](https://discuss.elastic.co/t/logstash-output-kafka-message-load-balancing/37804/3 "2015-12-28T23:18:39Z")

</div>

Logstash: 2.0.0-1.noarch.rpm  
Kafka: 2.11-0.9.0.0

According to the kafka doc, is it supposed to be round-robin if no message key is supplied?

Test A) I have script to covert the log message and then pipe them to the logstash.

input {  
stdin {  
codec =\> "json"  
}  
}

filter {  
date {  
match =\> ["timestamp", "YYYY-MM-dd HH:mm:ss,SSS"]  
locale =\> en  
}  
}

output {  
kafka {  
#broker\_list =\> "10.4.78.31:9092"  
bootstrap\_servers =\> "10.4.78.31:9092"  
acks =\> "0"  
topic\_id =\> "kafka-logs-new"  
message\_key =\> "%{timestamp}"  
}  
}

In this case, w/o message\_key specified, the massages are distributed across 3 partitions.

Test B) I have logstash installed on the cluster of servers. Following is snippet of logstash conf file:  
input {  
file {  
type =\> "zookeeper"  
path =\> "/home//logs/zookeeper.out._"  
exclude =\> "_.gz"  
}  
file {  
type =\> "cassandra"  
path =\> "/home//logs/cassandra/system\*"  
}  
.........  
}  
filter {  
.....  
grok {  
match =\> {  
"message" =\>"(?\<message\_hash\_key\>[\d\w\s\W]{30})"  
}  
}  
}  
output {  
kafka {  
bootstrap\_servers =\> "10.4.78.31:9092"  
topic\_id =\> "kafka-logs-new"  
#message\_key =\> "%{message\_hash\_key}"  
}  
}

since I can NOTt use "@timestamp" in "message\_key" in output, just parse  
first 30 character of log message, which contains time stamp plus other  
information.

In test B, all the messages go to the same partition. My goal is to distribute the logs among the 3 partitions so that 3 logstash servers can read the messages off 3 partitions.

Thanks

---

<div class="post-metadata">

**Author:** ![Lilit](https://avatars.discourse-cdn.com/v4/letter/l/df705f/32.png) [@Lilit](https://discuss.elastic.co/u/Lilit)\
**Post date:** [October 14, 2016, 8:46am UTC](https://discuss.elastic.co/t/logstash-output-kafka-message-load-balancing/37804/4 "2016-10-14T08:46:50Z")

</div>

try add field to the event by mutate plugin in filter, then assign new field to message\_key.  
In my case I had a json event.

input {  
file {  
path =\> "/usr/local/tomcat/log/event.log"  
#start\_position =\> beginning  
#ignore\_older =\> 0  
}  
}  
filter{  
json {  
source =\> "message"  
target =\> "message"  
}  
mutate {  
add\_field =\>{  
"user"=\> "%{[message][userId]}"  
}  
}  
}  
output {  
kafka {  
topic\_id =\> "topic-name"  
bootstrap\_servers =\> "server1:9092,server2:9092,server3:9092"  
message\_key =\> "%{user}"  
}  
}  
This code works!

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [July 6, 2017, 4:34am UTC](https://discuss.elastic.co/t/logstash-output-kafka-message-load-balancing/37804/5 "2017-07-06T04:34:14Z")

</div>