# Kafka AvroDesrializer - Logstash **URL:** \ **Category:** Logstash\ **Created:** [March 13, 2019, 5:09pm UTC](https://discuss.elastic.co/t/kafka-avrodesrializer-logstash/172189 "2019-03-13T17:09:31Z")\ **Posts on this page:** 6\ **Page:** 1
**Author:** ![DSPN](https://avatars.discourse-cdn.com/v4/letter/d/c89c15/32.png) [@DSPN](https://discuss.elastic.co/u/DSPN)\ **Post date:** [March 13, 2019, 5:09pm UTC](https://discuss.elastic.co/t/kafka-avrodesrializer-logstash/172189/1 "2019-03-13T17:09:31Z")
input { kafka{ group\_id =\> "group1" topics =\> ["Topic1"] bootstrap\_servers =\> "192.168.0.1:9092" codec =\> avro { schema\_uri =\> "=/logstash-6.6.0/avrotest.avsc" } key\_deserializer\_class =\> "io.confluent.kafka.serializers.KafkaAvroDeserializer" value\_deserializer\_class =\> "io.confluent.kafka.serializers.KafkaAvroDeserializer" } } output{ stdout{ codec =\> rubydebug } } This config is not working. Has anyone made logstash work with Avro serialized kafka messages. [2019-03-13T11:53:12,531][ERROR][logstash.inputs.kafka] Unable to create Kafka consumer from given configuration {:kafka\_error\_message=\>org.apache.kafka.common.config.ConfigException: Invalid value io.confluent.kafka.serializers.KafkaAvroDeserializer for configuration key.deserializer: Class io.confluent.kafka.serializers.KafkaAvroDeserializer could not be found., :cause=\>nil} [2019-03-13T11:53:12,532][ERROR][logstash.pipeline] A plugin had an unrecoverable error. Will restart this plugin. ---
**Author:** ![Chris\_Lyons](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/chris_lyons/32/48107_2.png) [@Chris\_Lyons](https://discuss.elastic.co/u/Chris_Lyons)\ **Post date:** [March 14, 2019, 1:42am UTC](https://discuss.elastic.co/t/kafka-avrodesrializer-logstash/172189/2 "2019-03-14T01:42:47Z")
This is an example of our config: ``` value_deserializer_class => "org.apache.kafka.common.serialization.ByteArrayDeserializer" codec => avro {schema_uri => "/etc/logstash/ancillary/avro/incident/.avsc"} ``` We don't need specific avro deserializer classes just need the ByteArrayDeserializer as it's stored in binary on the topic. All I care about is the value because the avro schema will define the key once in the pipeline. note: might take a look at your avro schema path, I don't think you need that '=' character at the start ---
**Author:** ![DSPN](https://avatars.discourse-cdn.com/v4/letter/d/c89c15/32.png) [@DSPN](https://discuss.elastic.co/u/DSPN)\ **Post date:** [March 14, 2019, 2:15am UTC](https://discuss.elastic.co/t/kafka-avrodesrializer-logstash/172189/3 "2019-03-14T02:15:47Z")
## Thanks Chris. Below is the error I'm receiving when i use only the avro codec in the input config without specifying any deserialization class [2019-03-13T13:29:51,596][FATAL][logstash.runner] An unexpected error occurred! {:error=\>#\, :backtrace=>["org/jruby/ext/stringio/StringIO.java:785:in`read'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/avro-1.8.2/lib/avro/io.rb:106:in `read'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/avro-1.8.2/lib/avro/io.rb:93:in`read\_bytes'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/avro-1.8.2/lib/avro/io.rb:99:in `read_string'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/avro-1.8.2/lib/avro/io.rb:299:in`read\_data'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/avro-1.8.2/lib/avro/io.rb:384:in `block in read_record'", "org/jruby/RubyArray.java:1734:in`each'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/avro-1.8.2/lib/avro/io.rb:382:in `read_record'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/avro-1.8.2/lib/avro/io.rb:310:in`read\_data'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/avro-1.8.2/lib/avro/io.rb:275:in `read'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/logstash-codec-avro-3.2.3-java/lib/logstash/codecs/avro.rb:77:in`decode'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/logstash-input-kafka-8.3.1/lib/logstash/inputs/kafka.rb:258:in `block in thread_runner'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/logstash-input-kafka-8.3.1/lib/logstash/inputs/kafka.rb:257:in`block in thread\_runner'"]} [2019-03-13T13:29:51,851][ERROR][org.logstash.Logstash] java.lang.IllegalStateException: Logstash stopped processing because of an error: (SystemExit) exit * * * Below is the error I'm receiving when i use the avro codec in the input config specifying "value\_deserializer\_class =\> "org.apache.kafka.common.serialization.ByteArrayDeserializer" as derserialization class [2019-03-13T21:03:02,763][FATAL][logstash.runner] An unexpected error occurred! {:error=\>#\, :backtrace=\>["org/jruby/ext/stringio/StringIO.java:788:in `read'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/avro-1.8.2/lib/avro/io.rb:106:in`read'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/avro-1.8.2/lib/avro/io.rb:93:in `read_bytes'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/avro-1.8.2/lib/avro/io.rb:99:in`read\_string'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/avro-1.8.2/lib/avro/io.rb:299:in `read_data'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/avro-1.8.2/lib/avro/io.rb:384:in`block in read\_record'", "org/jruby/RubyArray.java:1734:in `each'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/avro-1.8.2/lib/avro/io.rb:382:in`read\_record'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/avro-1.8.2/lib/avro/io.rb:310:in `read_data'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/avro-1.8.2/lib/avro/io.rb:275:in`read'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/logstash-codec-avro-3.2.3-java/lib/logstash/codecs/avro.rb:77:in `decode'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/logstash-input-kafka-8.3.1/lib/logstash/inputs/kafka.rb:258:in`block in thread\_runner'", "~/Documents/Elastic/logstash-6.6.0/vendor/bundle/jruby/2.3.0/gems/logstash-input-kafka-8.3.1/lib/logstash/inputs/kafka.rb:257:in `block in thread\_runner'"]} [2019-03-13T21:03:02,879][ERROR][org.logstash.Logstash] java.lang.IllegalStateException: Logstash stopped processing because of an error: (SystemExit) exit * * * _Background_- The kafka msgs Im trying to read are Avro serialized using the class "io.confluent.kafka.serializers.KafkaAvroSerializer". Is there any way I can use the confluent "io.confluent.kafka.serializers.KafkaAvroDeserializer" class in logstash. Has anyone implemented such a thing. ---
**Author:** ![Chris\_Lyons](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/chris_lyons/32/48107_2.png) [@Chris\_Lyons](https://discuss.elastic.co/u/Chris_Lyons)\ **Post date:** [March 18, 2019, 5:31pm UTC](https://discuss.elastic.co/t/kafka-avrodesrializer-logstash/172189/4 "2019-03-18T17:31:59Z")
I am not certain that the logstash plugin will support that serialization: [https://github.com/logstash-plugins/logstash-input-kafka/issues/37](https://github.com/logstash-plugins/logstash-input-kafka/issues/37) If there is a way to use the standard binary serialization on the producer side things should work, but beyond that you are probably going to have to customize the LS plugin. Unless someone else in the community has some ideas... ---
**Author:** ![DSPN](https://avatars.discourse-cdn.com/v4/letter/d/c89c15/32.png) [@DSPN](https://discuss.elastic.co/u/DSPN)\ **Post date:** [March 19, 2019, 6:10pm UTC](https://discuss.elastic.co/t/kafka-avrodesrializer-logstash/172189/5 "2019-03-19T18:10:32Z")
Below codec worked for me. Im posting this here incase anyone has a similar use case. > ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [April 16, 2019, 6:10pm UTC](https://discuss.elastic.co/t/kafka-avrodesrializer-logstash/172189/6 "2019-04-16T18:10:39Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.