# Error while diverting output to multiple secured kafka topic

**URL:** <https://discuss.elastic.co/t/error-while-diverting-output-to-multiple-secured-kafka-topic/196746>\
**Category:** Logstash\
**Created:** [August 26, 2019, 10:12am UTC](https://discuss.elastic.co/t/error-while-diverting-output-to-multiple-secured-kafka-topic/196746 "2019-08-26T10:12:53Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![amrutadale](https://avatars.discourse-cdn.com/v4/letter/a/71c47a/32.png) [@amrutadale](https://discuss.elastic.co/u/amrutadale)\
**Post date:** [August 26, 2019, 10:12am UTC](https://discuss.elastic.co/t/error-while-diverting-output-to-multiple-secured-kafka-topic/196746/1 "2019-08-26T10:12:53Z")

</div>

I am trying to divert output to different secured kafka topics based on the topic name received from HTTP end point.

for single topic my configuration is working fine. If I add **if else based on topic name** condition intermittently I keep on getting below error.

_working cofig_

input {  
http {  
host =\> "127.0.0.1" # default: 0.0.0.0  
port =\> "9600" # default: 8080  
codec =\> "json"  
}  
}

filter  
{

mutate { add\_field =\> { "[@metadata][topic1]" =\> "%{[headers][topic]}"} }  
mutate { add\_field =\> { "[@metadata][kafka\_key]" =\> "%{[headers][kafka\_key]}"} }  
mutate {remove\_field =\> [headers] }

}  
output {

if [@metadata][topic1] == "data\_cop\_test\_topic" {  
kafka {  
bootstrap\_servers =\> "xxx"  
topic\_id =\> "%{[@metadata][topic1]}"  
jaas\_path =\> ["C:\kafka\_kerberos\jass1.conf"]  
kerberos\_config =\> ["C:\kafka\_kerberos\krb5.conf"]  
sasl\_kerberos\_service\_name =\> "kafka"  
security\_protocol =\> "SASL\_PLAINTEXT"  
sasl\_mechanism =\> "GSSAPI"  
message\_key =\> "%{[@metadata][kafka\_key]}"  
codec =\> json{}  
}  
}

## }

**In the output if I add something like**

output {

if [@metadata][topic1] == "data\_cop\_test\_topic" {  
kafka {  
bootstrap\_servers =\> "xxx"  
topic\_id =\> "%{[@metadata][topic1]}"  
jaas\_path =\> ["C:\kafka\_kerberos\jass1.conf"]  
kerberos\_config =\> ["C:\kafka\_kerberos\krb5.conf"]  
sasl\_kerberos\_service\_name =\> "kafka"  
security\_protocol =\> "SASL\_PLAINTEXT"  
sasl\_mechanism =\> "GSSAPI"  
message\_key =\> "%{[@metadata][kafka\_key]}"  
codec =\> json{}  
}  
}

else if [@metadata][topic1] == "CDO\_TestTopic1" {  
kafka {  
bootstrap\_servers =\> "xxx"  
topic\_id =\> "%{[@metadata][topic1]}"  
jaas\_path =\> ["C:\kafka\_kerberos\jass2.conf"]  
kerberos\_config =\> ["C:\kafka\_kerberos\krb5.conf"]  
sasl\_kerberos\_service\_name =\> "kafka"  
security\_protocol =\> "SASL\_PLAINTEXT"  
sasl\_mechanism =\> "GSSAPI"  
message\_key =\> "%{[@metadata][kafka\_key]}"  
codec =\> json{}  
}  
}

}  
I start getting below error.

[2019-08-26T13:07:39,239][INFO][org.apache.kafka.common.utils.AppInfoParser] Kafka version : 2.1.0  
[2019-08-26T13:07:39,239][INFO][org.apache.kafka.common.utils.AppInfoParser] Kafka commitId : eec43959745f444f  
[2019-08-26T13:07:39,316][INFO][logstash.javapipeline] Starting pipeline {:pipeline\_id=\>"main", "pipeline.workers"=\>3, "pipeline.batch.size"=\>125, "pipeline.batch.delay"=\>50, "pipeline.max\_inflight"=\>375, :thread=\>"#\<Thread:0x49fef6e8 run\>"}  
[2019-08-26T13:07:39,973][INFO][logstash.javapipeline] Pipeline started {"pipeline.id"=\>"main"}  
[2019-08-26T13:07:40,020][INFO][logstash.inputs.http] Starting http input listener {:address=\>"127.0.0.1:9600", :ssl=\>"false"}  
[2019-08-26T13:07:40,083][INFO][logstash.agent] Pipelines running {:count=\>1, :running\_pipelines=\>[:main], :non\_running\_pipelines=\>}  
[2019-08-26T13:07:40,958][INFO][logstash.agent] Successfully started Logstash API endpoint {:port=\>9601}  
[2019-08-26T13:08:17,913][INFO][logstash.outputs.file] Opening file {:path=\>"C:/logstash-7.1.1/raw/2019.08.26.07.txt"}  
[2019-08-26T13:08:19,411][INFO][org.apache.kafka.clients.Metadata] Cluster ID: ZlmpEdElTL2NfyRvBn6ufA  
[2019-08-26T13:08:20,208][WARN][logstash.outputs.kafka] KafkaProducer.send() failed: org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [data\_cop\_test\_topic] {:exception=\>java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [data\_cop\_test\_topic]}  
[2019-08-26T13:08:20,224][INFO][logstash.outputs.kafka] Sending batch to Kafka failed. Will retry after a delay. {:batch\_size=\>1, :failures=\>1, :sleep=\>0.1}  
[2019-08-26T13:08:20,676][WARN][logstash.outputs.kafka] KafkaProducer.send() failed: org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [data\_cop\_test\_topic] {:exception=\>java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [data\_cop\_test\_topic]}  
[2019-08-26T13:08:20,676][INFO][logstash.outputs.kafka] Sending batch to Kafka failed. Will retry after a delay. {:batch\_size=\>1, :failures=\>1, :sleep=\>0.1}

Can someon please suggest if they had faced similar issue with multiple secured kafka topics?

---

<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:** [September 23, 2019, 10:12am UTC](https://discuss.elastic.co/t/error-while-diverting-output-to-multiple-secured-kafka-topic/196746/2 "2019-09-23T10:12:58Z")

</div>

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