# How to Ignore some messages from Kafka topic when consuming using Logstash

**URL:** <https://discuss.elastic.co/t/how-to-ignore-some-messages-from-kafka-topic-when-consuming-using-logstash/151505>\
**Category:** Logstash\
**Created:** [October 8, 2018, 5:59pm UTC](https://discuss.elastic.co/t/how-to-ignore-some-messages-from-kafka-topic-when-consuming-using-logstash/151505 "2018-10-08T17:59:31Z")\
**Posts on this page:** 9\
**Page:** 1

<div class="post-metadata">

**Author:** ![deeps](https://avatars.discourse-cdn.com/v4/letter/d/ccd318/32.png) [@deeps](https://discuss.elastic.co/u/deeps)\
**Post date:** [October 8, 2018, 5:59pm UTC](https://discuss.elastic.co/t/how-to-ignore-some-messages-from-kafka-topic-when-consuming-using-logstash/151505/1 "2018-10-08T17:59:32Z")

</div>

I have ten's of different messages in a kafka topic. I need to just consume messages that has only "America" in it. I have tried some suggestions available on the blog but no luck, could anyone help?

Flow is App -\> Kafka topic -\> Logstash consume pipeline and output to local ES.

Sample config:  
input {  
kafka {  
bootstrap\_servers =\> "broker:9093"  
security\_protocol =\> "SSL"  
ssl\_keystore\_location =\> "/etc/logstash/keystore.jks"  
ssl\_keystore\_password =\> "password"  
ssl\_truststore\_location =\> "/etc/logstash/truststore.jks"  
ssl\_truststore\_password =\> "password"  
topics =\> ["topic1"]  
codec =\> "json"  
}  
}  
filter {  
json{  
source =\> "message"  
}  
}  
output {  
if "America" in [message] {  
elasticsearch {  
hosts =\> localhost  
manage\_template =\> false  
index =\> "test"  
}  
}  
}

---

<div class="post-metadata">

**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\
**Post date:** [October 8, 2018, 7:35pm UTC](https://discuss.elastic.co/t/how-to-ignore-some-messages-from-kafka-topic-when-consuming-using-logstash/151505/2 "2018-10-08T19:35:15Z")

</div>

Show us an example unwanted document. Copy/paste the raw JSON from Kibana's JSON tab.

---

<div class="post-metadata">

**Author:** ![deeps](https://avatars.discourse-cdn.com/v4/letter/d/ccd318/32.png) [@deeps](https://discuss.elastic.co/u/deeps)\
**Post date:** [October 8, 2018, 7:38pm UTC](https://discuss.elastic.co/t/how-to-ignore-some-messages-from-kafka-topic-when-consuming-using-logstash/151505/3 "2018-10-08T19:38:49Z")

</div>

{  
"\_index": "test",  
"\_type": "doc",  
"\_id": "XX",  
"\_version": 1,  
"\_score": null,  
"\_source": {  
"meta": {  
"message\_type": "CREATE\_ACCOUNT",  
"create\_timestamp": "2018-10-02T19:04:12.416Z",  
"message\_send\_time": "2018-10-02T19:04:12.416Z",  
"message\_id": "XX"  
},  
"@version": "1",  
"@timestamp": "2018-10-08T19:05:26.335Z",  
"payload": {  
"email": "XX@XX.com",  
"last\_name": "XX",  
"first\_name": "XX",  
"value": XXOUT",  
"id": "XX"  
}  
},  
"fields": {  
"@timestamp": [  
"2018-10-08T19:05:26.335Z"  
],  
"meta.message\_send\_time": [  
"2018-10-02T19:04:12.416Z"  
],  
"meta.create\_timestamp": [  
"2018-10-02T19:04:12.416Z"  
]  
},  
"sort": [  
1539025526335  
]  
}

---

<div class="post-metadata">

**Author:** ![deeps](https://avatars.discourse-cdn.com/v4/letter/d/ccd318/32.png) [@deeps](https://discuss.elastic.co/u/deeps)\
**Post date:** [October 8, 2018, 7:49pm UTC](https://discuss.elastic.co/t/how-to-ignore-some-messages-from-kafka-topic-when-consuming-using-logstash/151505/4 "2018-10-08T19:49:39Z")

</div>

So I have different message\_type's. Ex: CREATE\_ACCOUNT, DELETE\_ACCOUNT  
I would like to only consume CREATE\_ACCOUNT.

---

<div class="post-metadata">

**Author:** ![deeps](https://avatars.discourse-cdn.com/v4/letter/d/ccd318/32.png) [@deeps](https://discuss.elastic.co/u/deeps)\
**Post date:** [October 9, 2018, 12:41am UTC](https://discuss.elastic.co/t/how-to-ignore-some-messages-from-kafka-topic-when-consuming-using-logstash/151505/5 "2018-10-09T00:41:00Z")

</div>

I was trying:

filter {  
if [@meta.message\_type] == "DELETE\_ACCOUNT" { drop{ } }  
}

but no luck, any help is appreciated! TIA.

---

<div class="post-metadata">

**Author:** ![deeps](https://avatars.discourse-cdn.com/v4/letter/d/ccd318/32.png) [@deeps](https://discuss.elastic.co/u/deeps)\
**Post date:** [October 9, 2018, 2:22pm UTC](https://discuss.elastic.co/t/how-to-ignore-some-messages-from-kafka-topic-when-consuming-using-logstash/151505/6 "2018-10-09T14:22:01Z")

</div>

hey @magnusbaeck could you help when you get chance?

---

<div class="post-metadata">

**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\
**Post date:** [October 9, 2018, 6:37pm UTC](https://discuss.elastic.co/t/how-to-ignore-some-messages-from-kafka-topic-when-consuming-using-logstash/151505/7 "2018-10-09T18:37:33Z")

</div>

> filter {  
> if [@meta.message\_type] == "DELETE\_ACCOUNT" { drop{ } }  
> }

Replace `[@meta.message_type]` with `[meta][message_type]`.

> **[Accessing event data and fields | Logstash Reference \[8.11\] | Elastic](https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html)**

---

<div class="post-metadata">

**Author:** ![deeps](https://avatars.discourse-cdn.com/v4/letter/d/ccd318/32.png) [@deeps](https://discuss.elastic.co/u/deeps)\
**Post date:** [October 9, 2018, 6:48pm UTC](https://discuss.elastic.co/t/how-to-ignore-some-messages-from-kafka-topic-when-consuming-using-logstash/151505/8 "2018-10-09T18:48:59Z")

</div>

that did the trick! thanks much! you are the best!! @magnusbaeck

---

<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:** [November 6, 2018, 6:48pm UTC](https://discuss.elastic.co/t/how-to-ignore-some-messages-from-kafka-topic-when-consuming-using-logstash/151505/9 "2018-11-06T18:48:59Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.