# Logstash Kafka Avro integration failing
**URL:** https://discuss.elastic.co/t/logstash-kafka-avro-integration-failing/192499
**Category:** Logstash
**Created:** [July 26, 2019, 9:13pm UTC](https://discuss.elastic.co/t/logstash-kafka-avro-integration-failing/192499 "2019-07-26T21:13:04Z")
**Posts on this page:** 7
**Page:** 1
### Author:  [@NSK](https://discuss.elastic.co/u/NSK)
#### Post date: [July 26, 2019, 9:13pm UTC](https://discuss.elastic.co/t/logstash-kafka-avro-integration-failing/192499/1 "2019-07-26T21:13:04Z")
Hi - I am trying to consume a topic from Kafka using Avro Deserializer in Logstash and getting the below error.
[ERROR][logstash.inputs.kafka] Unable to create Kafka consumer from given configuration {:kafka\_error\_message=\>org.apache.kafka.common.KafkaException: Failed to construct kafka consumer, :cause=\>io.confluent.common.config.ConfigException: Missing required configuration "schema.registry.url" which has no default value.}
[2019-07-26T16:58:22,736][ERROR][logstash.javapipeline] A plugin had an unrecoverable error. Will restart this plugin.
Pipeline\_id:main
I have provided avro\_uri in codec however, the settings is not been read by the logstash.
Here is my Logstash Config file
input {
kafka {
bootstrap\_servers =\> "kafka1:9911,kafka2:9911,kafka3:9911"
topics =\> "Elastic\_new"
#max\_poll\_records =\> "256"
auto\_offset\_reset =\> earliest
group\_id =\> "logstash104"
ssl\_truststore\_location =\>"/elasticsearch/logstash7.1.1/kafka\_files/kafka.client.truststore.jks"
ssl\_truststore\_password =\> "abcdef"
security\_protocol =\> "SSL"
#consumer\_threads =\> 10
key\_deserializer\_class =\> "io.confluent.kafka.serializers.KafkaAvroDeserializer"
value\_deserializer\_class =\> "io.confluent.kafka.serializers.KafkaAvroDeserializer"
#schema\_registry\_url =\> "[https://kafka:9990](https://kafka:9990)"
codec =\> avro {
schema\_uri =\> "/elasticsearch/logstash-7.1.1/kafka\_files/ticketInfo.avsc"
tag\_on\_failure =\> true
#register\_schema =\> true
#base64\_encoding =\> false
}
}
}
output {
elasticsearch {
index =\> "topic\_es2"
document\_id =\> "%{tktnum}%"
action =\> "update"
doc\_as\_upsert =\> "true"
retry\_on\_conflict =\> 5
hosts =\> ["npes1:9200"]
}
stdout { codec =\> rubydebug }
}
---
### Author:  [@NSK](https://discuss.elastic.co/u/NSK)
#### Post date: [July 27, 2019, 2:08pm UTC](https://discuss.elastic.co/t/logstash-kafka-avro-integration-failing/192499/2 "2019-07-27T14:08:52Z")
Can someone please help with the issue?
---
### Author:  [@NSK](https://discuss.elastic.co/u/NSK)
#### Post date: [July 29, 2019, 11:50pm UTC](https://discuss.elastic.co/t/logstash-kafka-avro-integration-failing/192499/3 "2019-07-29T23:50:06Z")
Hi Everyone,
Here are the jars that i added in logstash to consume data from kafka.
kafka-avro-serializer-5.0.0.jar
kafka-clients-2.0.0.jar
common-config-5.0.0.jar
common-utils-5.0.0.jar
kafka-schema-registry-client-5.0.0.jar
avro-1.8.2.jar
avro-maven-plugin-1.8.2.jar
avro-compiler-1.8.2.jar
jackson-mapper-asl-1.9.13.jar
jackson-core-asl-1.9.13.jar
Is there an issue with the plugins that am using to consume data from Kafka using avro deserializer?
---
### Author:  [@guyboertje](https://discuss.elastic.co/u/guyboertje)
#### Post date: [July 30, 2019, 3:32pm UTC](https://discuss.elastic.co/t/logstash-kafka-avro-integration-failing/192499/4 "2019-07-30T15:32:46Z")
I don't think you can use both `value_deserializer_class => "io.confluent.kafka.serializers.KafkaAvroDeserializer"` and the avro codec. Decoding twice?
As we don't have a setting that passes a value to the `schema.registry.url` property of the Java Kafka Client you will have to use the avro codec only.
---
### Author:  [@NSK](https://discuss.elastic.co/u/NSK)
#### Post date: [July 30, 2019, 3:58pm UTC](https://discuss.elastic.co/t/logstash-kafka-avro-integration-failing/192499/5 "2019-07-30T15:58:28Z")
Thanks for the response. I have removed the deserializer class from the conf file as stated:
input {
kafka {
bootstrap\_servers =\> "kafka1:9911,kafka2:9911,kafka3:9911"
topics =\> "Elastic\_new"
#max\_poll\_records =\> "256"
auto\_offset\_reset =\> earliest
group\_id =\> "logstash104"
ssl\_truststore\_location =\>"/elasticsearch/logstash7.1.1/kafka\_files/kafka.client.truststore.jks"
ssl\_truststore\_password =\> "abcdef"
security\_protocol =\> "SSL"
codec =\> avro {
schema\_uri =\> "/elasticsearch/logstash-7.1.1/kafka\_files/ticketInfo.avsc"
tag\_on\_failure =\> true
}
}
}
This means the conf will be using the default string deserializer.
When i execute the script, got below output.
{
"ticketObjects" =\> ,
"@version" =\> "1",
"@timestamp" =\> 2019-07-30T15:51:07.676Z
}
---
### Author:  [@guyboertje](https://discuss.elastic.co/u/guyboertje)
#### Post date: [July 30, 2019, 4:28pm UTC](https://discuss.elastic.co/t/logstash-kafka-avro-integration-failing/192499/6 "2019-07-30T16:28:03Z")
See [this StackOverflow answer](https://stackoverflow.com/a/42105883)
You need to use the ByteArraySerializer in Kafka to preserve the binary data.
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [August 27, 2019, 4:28pm UTC](https://discuss.elastic.co/t/logstash-kafka-avro-integration-failing/192499/7 "2019-08-27T16:28:03Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.