# Logstash Kafka Input Plugin Data has Type Already **URL:** \ **Category:** Logstash\ **Created:** [August 3, 2016, 11:28pm UTC](https://discuss.elastic.co/t/logstash-kafka-input-plugin-data-has-type-already/57161 "2016-08-03T23:28:19Z")\ **Posts on this page:** 7\ **Page:** 1
**Author:** ![kingston](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/kingston/32/11196_2.png) [@kingston](https://discuss.elastic.co/u/kingston)\ **Post date:** [August 3, 2016, 11:28pm UTC](https://discuss.elastic.co/t/logstash-kafka-input-plugin-data-has-type-already/57161/1 "2016-08-03T23:28:20Z")
I'm using the Kafka Input plugin and noticed that my elasticsearch indices were being named incorrectly. After debugging, I realized that the kafka topic had documents that already included a type field. Input.conf ``` input { kafka { type => "kafka" topic_id => "topic" zk_connect => "zk1, zk2, zk3" } } ``` Data coming in may look something like: ``` {"date_created":"2016-08-03 16:19:18", "unix_time":1470266358, "date_short":"2016-08-03", "local_day_of_week":"Wed", "local_time":"1619", "type":2, "event_type":"topic"} ``` So, if you notice that **type** field has a value of **2**. When I look on **[http://elasticsearch/\_cat/indices](http://elasticsearch/_cat/indices)** it just has **green open 2-2016.08.03** instead of **topic-2016.08.03**. Are there any ways around this? ---
**Author:** ![Joe\_Lawson](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/joe_lawson/32/3390_2.png) [@Joe\_Lawson](https://discuss.elastic.co/u/Joe_Lawson)\ **Post date:** [August 4, 2016, 2:08am UTC](https://discuss.elastic.co/t/logstash-kafka-input-plugin-data-has-type-already/57161/2 "2016-08-04T02:08:37Z")
Can you share the Elasticsearch output config? ---
**Author:** ![kingston](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/kingston/32/11196_2.png) [@kingston](https://discuss.elastic.co/u/kingston)\ **Post date:** [August 4, 2016, 5:25pm UTC](https://discuss.elastic.co/t/logstash-kafka-input-plugin-data-has-type-already/57161/3 "2016-08-04T17:25:03Z")
Hey Joe, Here's a (hostname modified) version of my output config. ``` output { elasticsearch { hosts => ["elasticsearchserver"] index => "%{type}-%{+yyyy.MM.dd}" workers => 1 } } ``` Thanks! ---
**Author:** ![Joe\_Lawson](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/joe_lawson/32/3390_2.png) [@Joe\_Lawson](https://discuss.elastic.co/u/Joe_Lawson)\ **Post date:** [August 4, 2016, 8:13pm UTC](https://discuss.elastic.co/t/logstash-kafka-input-plugin-data-has-type-already/57161/4 "2016-08-04T20:13:15Z")
If you cannot change the message from having a "type" field already I would just try using another field ie [https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-add\_field](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-add_field) ```auto input { kafka { add_field => { "log_origin" => "kafka" } ... } ``` ```auto output { elasticsearch { hosts => ["elasticsearchserver"] index => "%{log_origin}-%{+yyyy.MM.dd}" workers => 1 } } ``` ---
**Author:** ![kingston](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/kingston/32/11196_2.png) [@kingston](https://discuss.elastic.co/u/kingston)\ **Post date:** [August 4, 2016, 9:02pm UTC](https://discuss.elastic.co/t/logstash-kafka-input-plugin-data-has-type-already/57161/5 "2016-08-04T21:02:13Z")
Thanks Joe! I think this will work for my use case. Kingston ---
**Author:** ![moni15moni](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/moni15moni/32/42311_2.png) [@moni15moni](https://discuss.elastic.co/u/moni15moni)\ **Post date:** [February 14, 2017, 6:01am UTC](https://discuss.elastic.co/t/logstash-kafka-input-plugin-data-has-type-already/57161/6 "2017-02-14T06:01:34Z")
Hi Joe, I have a same use case like my message format as follows from kafka output is , "rowid":0,"sentinel\_id":4413,"cust\_id":1,"devi\_id":5309,"first\_occurance":"2017-02-13 [05:30:14","type":6,"action":25,"protocol":null,"src":3158395560,"dst":2094651427,"src\_name":null,"dst\_name":null,"dst\_port":110,"user":"Edward","user\_group":"HR","policy\_name":null,"misc\_1":null,"misc\_2":"port3Edward@example.com](mailto:05:30:14%22,%22type%22:6,%22action%22:25,%22protocol%22:null,%22src%22:3158395560,%22dst%22:2094651427,%22src_name%22:null,%22dst_name%22:null,%22dst_port%22:110,%22user%22:%22Edward%22,%22user_group%22:%22HR%22,%22policy_name%22:null,%22misc_1%22:null,%22misc_2%22:%22port3Edward@example.com)","misc\_3":"port2Edward@exampls.com","misc\_8":null,"misc\_9":null,"rawlog":"\<723\>Feb 13 05:30:13 10.2.1.224 date=2017-02-13 time=05:30:13 devname=FGCorp001 device\_id=FGT8004271490115 log\_id=050926726 type=emailfilter subtype=POP3 pri=emergency fwver=040004 policyid=78 serial=19286623 user="Edward" group="HR" vd="root" src=188.65.74.168 sport=110 src\_port=110 src\_int="port3" dst=124.217.216.35 dport=110 dst\_port=110 dst\_int"port2" service="110/pop3" carrier\_ep="EndPoint" profile="profile" status="blocked" [from="port3Edward@example.com](mailto:from=%22port3Edward@example.com)" [to="port2Edward@exampls.com](mailto:to=%22port2Edward@exampls.com)" tracker="Tracker" msg="from email address is in email blacklist."","rawlog\_hash":8789335990287780976,"evtcount":1,"inbytes":0,"outbytes":0,"totalbytes":0} How can i check the condition for specific field ilike , if([dst]==124.217.216.35){ } condition,For my case this one is not giving expected output ---
**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:28am UTC](https://discuss.elastic.co/t/logstash-kafka-input-plugin-data-has-type-already/57161/7 "2017-07-06T04:28:39Z")