# Send different event to multiple output plugins in same config

**URL:** <https://discuss.elastic.co/t/send-different-event-to-multiple-output-plugins-in-same-config/307390>\
**Category:** Logstash\
**Created:** [June 16, 2022, 11:03am UTC](https://discuss.elastic.co/t/send-different-event-to-multiple-output-plugins-in-same-config/307390 "2022-06-16T11:03:40Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![priyankamondalhcl](https://avatars.discourse-cdn.com/v4/letter/p/57b2e6/32.png) [@priyankamondalhcl](https://discuss.elastic.co/u/priyankamondalhcl)\
**Post date:** [June 16, 2022, 11:03am UTC](https://discuss.elastic.co/t/send-different-event-to-multiple-output-plugins-in-same-config/307390/1 "2022-06-16T11:03:40Z")

</div>

Hi Team,

I have one pipeline with multiple output, elasticsearch and kafka.  
Now, the requirement is to sent a field to elasticsearch, but while sending to kafka that field needs to drop.  
Here my config says input is from elasticsearch and update one field from that set of data and again save it to elasticsearch and send to kafka. But I am not able to delete that field([customcol][flag]) while sending to kafka.  
\<  
elasticsearch {  
hosts =\> ["[https://10.1.1.X](https://10.1.1.X)  
:9200","[https://10.1.2.X:9200](https://10.1.2.X:9200)","[https://10.1.3.X:9200](https://10.1.3.X:9200)"]  
index =\> "iece\*"  
user =\> "myuser"  
password =\> "mypass"  
query =\> '{"query": { "bool": { "filter": [{ "bool": { "must\_not": [ { "match\_phrase": { "customcol.flag": "read" } }], "minimum\_should\_match": 1 } }, { "range": { "createdon": { "format": "strict\_date\_optional\_time", "gte": "now-1d", "lte": "now" } } } ] } }, "sort": [{ "createdon": { "order": "desc", "unmapped\_type": "boolean" } }] }'  
size =\> 500  
scroll =\> "5m"  
docinfo =\> true  
docinfo\_target =\> "[@metadata][doc]"  
schedule =\> "/10 \* \* \* \* \*"  
}  
}

filter{  
mutate {update =\> { "[customcol][flag]" =\> "read"}}

mutate {remove\_field =\> ["@version","process","tags", "splitby\_copy","flag", "@timestamp"] }

}  
output {  
elasticsearch {  
hosts =\> ["[https://10.1.2.X:9200](https://10.1.2.X:9200)","[https://10.1.4.X:9200](https://10.1.4.X:9200)","[https://10.1.3.X:9200](https://10.1.3.X:9200)"]  
index =\> "%{[@metadata][doc][\_index]}"  
document\_id =\> "%{[@metadata][doc][\_id]}"  
action =\> "update"  
doc\_as\_upsert =\> true  
manage\_template =\> true  
user =\> "myuser"  
password =\> "mypass"  
}

stdout { codec =\> rubydebug { metadata =\> true }}  
}  
output {  
kafka {  
bootstrap\_servers =\> "10.1.X.X:9092,10.2.X.X:9092,10.3.X.X:9092"  
topic\_id =\> "envdbuat\_allevents\_%{toolcustomername}\_%{toolmanager}"  
codec =\> json  
security\_protocol =\> "SSL"  
ssl\_endpoint\_identification\_algorithm =\> ""  
ssl\_key\_password =\> "abcdef"  
ssl\_keystore\_location =\> "/etc/logstash/conf.d/extraconf/lib/kafka\_certi\_new/kafka.client.keystore.jks"  
ssl\_keystore\_password =\> "abcdef"  
ssl\_truststore\_location =\> "/etc/logstash/conf.d/extraconf/lib/kafka\_certi\_new/kafka.client.keystore.jks"  
ssl\_truststore\_password =\> "abcdef"

}

}

>

---

<div class="post-metadata">

**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\
**Post date:** [June 16, 2022, 3:45pm UTC](https://discuss.elastic.co/t/send-different-event-to-multiple-output-plugins-in-same-config/307390/2 "2022-06-16T15:45:23Z")

</div>

> [@priyankamondalhcl](#):
>
> Now, the requirement is to sent a field to Elasticsearch, but while sending to kafka that field needs to drop.

Use pipeline-to-pipeline communication. See the [forked-path](https://www.elastic.co/guide/en/logstash/current/pipeline-to-pipeline.html#forked-path-pattern) pattern.

---

<div class="post-metadata">

**Author:** ![priyankamondalhcl](https://avatars.discourse-cdn.com/v4/letter/p/57b2e6/32.png) [@priyankamondalhcl](https://discuss.elastic.co/u/priyankamondalhcl)\
**Post date:** [June 20, 2022, 6:45am UTC](https://discuss.elastic.co/t/send-different-event-to-multiple-output-plugins-in-same-config/307390/3 "2022-06-20T06:45:33Z")

</div>

Thanks @Badger for your reply!

I solved it by calling API to update in Elasticsearch and remove that flag field from the event and then send it to kafka. below is my config:

input {  
Elasticsearch {  
hosts =\> ["[https://10.1.2.X:9200](https://10.1.2.X:9200)","[https://10.1.4.X:9200](https://10.1.4.X:9200)","[https://10.1.3.X:9200](https://10.1.3.X:9200)"]  
index =\> "iece\*"  
user =\> "myuser"  
password =\> "mypass"  
query =\> '{"query": { "bool": { "filter": [{ "bool": { "must\_not": [ { "match\_phrase": { "flag": "read" } }], "minimum\_should\_match": 1 } }, { "range": { "createdon": { "format": "strict\_date\_optional\_time", "gte": "now-1d", "lte": "now" } } } ] } }, "sort": [{ "createdon": { "order": "desc", "unmapped\_type": "boolean" } }] }'  
size =\> 500  
scroll =\> "5m"  
docinfo =\> true  
docinfo\_target =\> "[@metadata][doc]"  
schedule =\> "/10 \* \* \* \* \*"  
}  
}

filter{

ruby { code =\> "event.set('[@metadata][currentdate]', Time.now.strftime('%d-%m-%Y'))" }  
mutate { add\_field =\> { "[@metadata][indextime]" =\> "iece%{[@metadata][currentdate]}" } }

mutate {remove\_field =\> ["@version","process","tags", "splitby\_copy","flag", "@timestamp"] }

}  
filter {  
http {  
verb =\> "POST"  
url =\> "[https://elkurl:9200/%{[@metadata][doc][\_index]}/\_update/%{[@metadata][doc][\_id]}](https://elkurl:9200/%25%7B%5B@metadata%5D%5Bdoc%5D%5B_index%5D%7D/_update/%25%7B%5B@metadata%5D%5Bdoc%5D%5B_id%5D%7D)"  
user =\> "myuser"  
password =\> "mypass"  
body\_format =\> "json"  
body =\> '{"doc": {"flag": "read"}}'  
ecs\_compatibility =\> "disabled"  
}

mutate {remove\_field =\> ["body", "headers"] }  
}

output {

stdout { codec =\> rubydebug { metadata =\> true }}  
}  
output {  
kafka {  
bootstrap\_servers =\> "10.1.X.X:9092,10.2.X.X:9092,10.3.X.X:9092"  
topic\_id =\> "envdbuat\_allevents\_%{toolcustomername}\_%{toolmanager}"  
codec =\> json  
security\_protocol =\> "SSL"  
ssl\_endpoint\_identification\_algorithm =\> ""  
ssl\_key\_password =\> "abcdef"  
ssl\_keystore\_location =\> "/etc/logstash/conf.d/extraconf/lib/kafka\_certi\_new/kafka.client.keystore.jks"  
ssl\_keystore\_password =\> "abcdef"  
ssl\_truststore\_location =\> "/etc/logstash/conf.d/extraconf/lib/kafka\_certi\_new/kafka.client.keystore.jks"  
ssl\_truststore\_password =\> "abcdef"

}

}

---

<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 18, 2022, 6:45am UTC](https://discuss.elastic.co/t/send-different-event-to-multiple-output-plugins-in-same-config/307390/4 "2022-07-18T06:45:54Z")

</div>

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