Logstash with Kafka: unable to decode avro

Question

I am trying to use serialized avro events from the Kafka queue. The kafka queue is populated using a simple Java maker. For clarity, I separate three components:

Avro Scheme File

{"namespace": "example.avro", "type": "record", "name": "User", "fields": [ {"name": "name", "type": "string"}, {"name": "favorite_number", "type": ["int", "null"]}, {"name": "favorite_color", "type": ["string", "null"]} ] } 

Java Producer code snippet (User.class is created using avro-tools)

  User user1 = new User(); user1.setName("Alyssa"); user1.setFavoriteNumber(256); user1.setFavoriteColor("blue"); String topic = "MemoryTest"; // Properties set in 'props' KafkaProducer<Message, byte[]> producer = new KafkaProducer<Message, byte[]>(props); ByteArrayOutputStream out = new ByteArrayOutputStream(); DatumWriter<User> writer = new SpecificDatumWriter<User>(User.class); Encoder encoder = EncoderFactory.get().binaryEncoder(out, null); writer.write(user1, encoder); encoder.flush(); out.close(); byte[] serializedBytes = out.toByteArray(); producer.send(new ProducerRecord<Message, byte[]>(topic, serializedBytes)); 

Logstash Configuration File

 input { kafka { zk_connect => "localhost:2181" topic_id => "MemoryTest" type => "standard_event" group_id => "butiline_dash_prod" reset_beginning => true auto_offset_reset => smallest codec => { avro => { schema_uri => "/opt/ELK/logstash-1.5.4/bin/user.avsc" } } } } output { stdout { codec => rubydebug } } 

Problem

Logstash pipeline failure. When a new event is inserted into Kafka, I follow the logstash console:

 Alyssa blue {:exception=>#<NoMethodError: undefined method `decode' for ["avro", {"schema_uri"=>"/opt/ELK/logstash-1.5.4/bin/user.avsc"}]:Array>, :backtrace=>["/opt/ELK/logstash-1.5.4/vendor/bundle/jruby/1.9/gems/logstash-input-kafka-1.0.0/lib/logstash/inputs/kafka.rb:169:in `queue_event'", "/opt/ELK/logstash-1.5.4/vendor/bundle/jruby/1.9/gems/logstash-input-kafka-1.0.0/lib/logstash/inputs/kafka.rb:139:in `run'", "/opt/ELK/logstash-1.5.4/vendor/bundle/jruby/1.9/gems/logstash-core-1.5.4-java/lib/logstash/pipeline.rb:177:in `inputworker'", "/opt/ELK/logstash-1.5.4/vendor/bundle/jruby/1.9/gems/logstash-core-1.5.4-java/lib/logstash/pipeline.rb:171:in `start_input'"], :level=>:error} 

Answer 1

Finally I found out the error. Instead (as suggested on the Logstash website - https://www.elastic.co/guide/en/logstash/current/plugins-codecs-avro.html )

 codec => { avro => { schema_uri => "/opt/ELK/logstash-1.5.4/bin/user.avsc" } } 

The correct syntax (as suggested in the plugin documentation https://github.com/logstash-plugins/logstash-codec-avro/blob/master/DEVELOPER.md ):

 codec => avro { schema_uri => "/opt/ELK/logstash-1.5.4/bin/user.avsc" } 

I assume the syntax is changed.