# Send logs to kafka with different topic using zeek-kafka plugin

**URL:** <https://community.zeek.org/t/send-logs-to-kafka-with-different-topic-using-zeek-kafka-plugin/6812>\
**Category:** Zeek\
**Tags:** development\
**Created:** [November 26, 2022, 12:32pm UTC](https://community.zeek.org/t/send-logs-to-kafka-with-different-topic-using-zeek-kafka-plugin/6812 "2022-11-26T12:32:34Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![frank](https://yyz1.discourse-cdn.com/flex011/user_avatar/community.zeek.org/frank/32/637_2.png) [@frank](https://community.zeek.org/u/frank)\
**Post date:** [November 26, 2022, 12:32pm UTC](https://community.zeek.org/t/send-logs-to-kafka-with-different-topic-using-zeek-kafka-plugin/6812/1 "2022-11-26T12:32:34Z")

</div>

I want to send some modules’ log to kafka with different topic, it does’t work  
and I don’t see any errors output

4 redef Kafka::logs\_to\_send = set(HTTP::LOG, DNS::LOG);  
5 redef Kafka::topic\_name = “”;  
6 #redef Kafka::send\_all\_active\_logs = T;  
7 #redef Kafka::tag\_json = T;  
8  
9 redef Kafka::kafka\_conf = table(  
10 [“metadata.broker.list”] = “192.168.31.138:9092”  
11 );

33 event zeek\_init() &priority=-10  
34 {  
35 for (stream\_id in Log::active\_streams)  
36 {  
37 if (send\_to\_kafka(stream\_id))  
38 {  
39 local filter: Log::Filter = [  
40 $name = fmt(“kafka-%s”, stream\_id),  
41 $writer = Log::WRITER\_KAFKAWRITER,  
42 $config = table(  
43 [“metadata.broker.list”] = “192.168.31.138:9092”  
44 ),  
45 $path = fmt(“zeek\_%s”, stream\_id)  
46 ];  
47 print “----”,fmt(“zeek-%s”, stream\_id);  
48 Log::add\_filter(stream\_id, filter);  
49 }  
50 }  
51 }

---

<div class="post-metadata">

**Author:** ![awelzel](https://yyz1.discourse-cdn.com/flex011/user_avatar/community.zeek.org/awelzel/32/609_2.png) [@awelzel](https://community.zeek.org/u/awelzel)\
**Post date:** [November 28, 2022, 5:21pm UTC](https://community.zeek.org/t/send-logs-to-kafka-with-different-topic-using-zeek-kafka-plugin/6812/2 "2022-11-28T17:21:32Z")

</div>

> [@frank](#):
>
> 33 event zeek\_init() &priority=-10

Without having tried, could you attempt to use `&priority=-20` on your `zeek_init()` handler? The `zeek_init()` handler of the [zeek-kafka plugin](https://github.com/SeisoLLC/zeek-kafka/blob/b632187c7afdc75365a256a7d9bb2e5cc81ab46f/scripts/Seiso/Kafka/logs-to-kafka.zeek#L39) runs at `priority=-10` as well and my replace the filters you created. Using a lower priority guarantees your `zeek_init()` implementations runs later.

Hope this helps,  
Arne

---

<div class="post-metadata">

**Author:** ![frank](https://yyz1.discourse-cdn.com/flex011/user_avatar/community.zeek.org/frank/32/637_2.png) [@frank](https://community.zeek.org/u/frank)\
**Post date:** [November 29, 2022, 12:51am UTC](https://community.zeek.org/t/send-logs-to-kafka-with-different-topic-using-zeek-kafka-plugin/6812/3 "2022-11-29T00:51:44Z")

</div>

5 redef Kafka::topic\_name = “”;  
6 redef Kafka::send\_all\_active\_logs = T;  
7 #redef Kafka::tag\_json = T;  
8  
9 redef Kafka::kafka\_conf = table(  
10 [“metadata.broker.list”] = “192.168.31.138:9092”  
11 );

33 event zeek\_init() &priority=-10  
34 {  
35 for (stream\_id in Log::active\_streams)  
36 {  
37 if (send\_to\_kafka(stream\_id))  
38 {  
39 local topicName:string = “zeek”;  
40 local streamName:string = fmt(“%s”, stream\_id);  
41 local s\_t = split\_string(streamName, /::/);  
42 for( index\_st in s\_t)  
43 {  
44 topicName += “\_”;  
45 topicName +=s\_t[index\_st];  
46 }  
47 #print “–topic name—”, topicName, "stram id ", streamName;  
48 local filter: Log::Filter = [  
49 $name = fmt(“kafka-%s”, stream\_id),  
50 $writer = Log::WRITER\_KAFKAWRITER,  
51 $config = table(  
52 [“metadata.broker.list”] = “192.168.31.138:9092”  
53 ),  
54 $path = topicName  
55 ];  
56 Log::add\_filter(stream\_id, filter);  
57 }  
58 }  
}

this can works,  
seems Kafka::kafka\_conf must be configured.  
and topicName must be a string without “::”, stream\_id is always contains “::”