# Not able to read Kafka - Avro Schema messages **URL:** \ **Category:** Logstash\ **Created:** [July 6, 2015, 7:19pm UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979 "2015-07-06T19:19:56Z")\ **Posts on this page:** 17\ **Page:** 1
**Author:** ![karthikr](https://avatars.discourse-cdn.com/v4/letter/k/eada6e/32.png) [@karthikr](https://discuss.elastic.co/u/karthikr)\ **Post date:** [July 6, 2015, 7:19pm UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/1 "2015-07-06T19:19:56Z")
I want to read snappy compressed avro messages stored in kafka into elastic. I am using the Kafka Input Plugin and the Avro Logstash Codec but not able to read the compressed messages. Do note that I am able to read plain non avro schematic messages from Kafka. ---
**Author:** ![eperry](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/eperry/32/551_2.png) [@eperry](https://discuss.elastic.co/u/eperry)\ **Post date:** [July 7, 2015, 2:04pm UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/2 "2015-07-07T14:04:47Z")
Sounds to me like you need a Snappy / Avro codec plugin to decode the data after it has been pulled from the queue. ---
**Author:** ![karthikr](https://avatars.discourse-cdn.com/v4/letter/k/eada6e/32.png) [@karthikr](https://discuss.elastic.co/u/karthikr)\ **Post date:** [July 7, 2015, 2:19pm UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/3 "2015-07-07T14:19:09Z")
I am already using a AVRO plugin. ---
**Author:** ![eperry](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/eperry/32/551_2.png) [@eperry](https://discuss.elastic.co/u/eperry)\ **Post date:** [July 7, 2015, 2:31pm UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/4 "2015-07-07T14:31:42Z")
Sorry about that I looked at the Doc site and did not see it listed. After reading the plugin and looking at the Avro Ruby Library Compression is not implemented yet Filed in March AvroAVRO-1659 Add Snappy support for Ruby, it has not been closed yet. [https://issues.apache.org/jira/browse/AVRO-1659](https://issues.apache.org/jira/browse/AVRO-1659) ---
**Author:** ![karthikr](https://avatars.discourse-cdn.com/v4/letter/k/eada6e/32.png) [@karthikr](https://discuss.elastic.co/u/karthikr)\ **Post date:** [July 7, 2015, 2:40pm UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/5 "2015-07-07T14:40:53Z")
hey, thanks a lot! ---
**Author:** ![talevy](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/talevy/32/44896_2.png) [@talevy](https://discuss.elastic.co/u/talevy)\ **Post date:** [July 7, 2015, 8:36pm UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/6 "2015-07-07T20:36:26Z")
Ah yes, Snappy is not supported just yet. but please file an issue on the github for the avro codec. We may be able to explore a Java implementation, thereby replacing the ruby avro library with one that does support Snappy. ---
**Author:** ![karthikr](https://avatars.discourse-cdn.com/v4/letter/k/eada6e/32.png) [@karthikr](https://discuss.elastic.co/u/karthikr)\ **Post date:** [July 7, 2015, 9:59pm UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/7 "2015-07-07T21:59:43Z")
I removed the snappy compression to null (no) compression today. Even then the avro codec is not able to read the avsc and decode the file. Do note that we are using the KafkaAvroSerializer to read the data. ---
**Author:** ![talevy](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/talevy/32/44896_2.png) [@talevy](https://discuss.elastic.co/u/talevy)\ **Post date:** [July 8, 2015, 3:27pm UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/8 "2015-07-08T15:27:06Z")
> [@karthikr](#): > > KafkaAvroSerializer The Kafka Avro Serializer is not necessarily only Avro. From what I see it has some prepended values that are not avro encoded: > ---
**Author:** ![upendra](https://avatars.discourse-cdn.com/v4/letter/u/e68b1a/32.png) [@upendra](https://discuss.elastic.co/u/upendra)\ **Post date:** [June 8, 2016, 1:21pm UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/9 "2016-06-08T13:21:40Z")
Hi, Any solution to this problem??? We are also facing the same issue. We are trying to send messages from logstash to KAFKA to HDFS. We are using the following tech stack: 1. Logstash 2.3 - Current production version 2. Confluent 3.0. 3. Plugins: a. Logstash-kafka-Output plugin b. Logstash-codec-avro. 4. zookeeper: 3.4.6 5. KAFKA: 0.10.0.0 Our Logstash config file looks like this: input { stdin{} } filter { mutate { remove\_field =\> ["@timestamp","@version"] } } output { kafka { topic\_id =\> 'logstash\_logs14' codec =\> avro { schema\_uri =\> "/opt/logstash/bin/schema.avsc" } } } The schema.avsc file looks like this: { "type":"record", "name":"myrecord", "fields":[ {"name":"message","type":"string"}, {"name":"host","type":"string"} ] } 1. Start Zookeeper in its own terminal ./bin/zookeeper-server-start ./etc/kafka/zookeeper.properties 2 Start Kafka in its own terminal ./bin/kafka-server-start ./etc/kafka/server.properties 3 Start schema registry in its own terminal ./bin/schema-registry-start ./etc/schema-registry/schema-registry.properties 4 From logstash directory, run the following command bin/logstash -f ./bin/logstash.conf 5 Type the log message that you wish to send to kafka after running above command ex: "Hello World" 6 Consume the topic from Kafka ./bin/kafka-avro-console-consumer --zookeeper localhost:2181 --topic logstash\_logs14 --from-beginning _While consuming we get the following error:_ SLF4J: Class path contains multiple SLF4J bindings. SLF4J: Found binding in [jar:file:/root/confluent-3.0.0/share/java/kafka-serde-tools/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class] SLF4J: Found binding in [jar:file:/root/confluent-3.0.0/share/java/confluent-common/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class] SLF4J: Found binding in [jar:file:/root/confluent-3.0.0/share/java/schema-registry/slf4j-log4j12-1.7.6.jar!/org/slf4j/impl/StaticLoggerBinder.class] SLF4J: See [http://www.slf4j.org/codes.html#multiple\_bindings](http://www.slf4j.org/codes.html#multiple_bindings) for an explanation. SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory] Processed a total of 1 messages [2016-06-08 18:42:41,627] ERROR Unknown error when running consumer: (kafka.tools.ConsoleConsumer$:103) org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1 Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte! [2016-06-08 18:42:41,627] ERROR Unknown error when running consumer: (kafka.tools.ConsoleConsumer$:103) org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1 Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte! Thanks, Upendra ---
**Author:** ![Joe\_Lawson](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/joe_lawson/32/3390_2.png) [@Joe\_Lawson](https://discuss.elastic.co/u/Joe_Lawson)\ **Post date:** [June 9, 2016, 3:35am UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/10 "2016-06-09T03:35:14Z")
What versions are your Logstash Kafka plugins? I think 2.3 only supports Kafka 0.9.x ---
**Author:** ![upendra](https://avatars.discourse-cdn.com/v4/letter/u/e68b1a/32.png) [@upendra](https://discuss.elastic.co/u/upendra)\ **Post date:** [June 9, 2016, 2:01pm UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/11 "2016-06-09T14:01:24Z")
Hi Joe, Thanks for your response. We have updated logstash to 2.3 but kafka is 0.10 (latest version). But still we are getting same issue. Should we downgrade Kafka to 0.9. ---
**Author:** ![Joe\_Lawson](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/joe_lawson/32/3390_2.png) [@Joe\_Lawson](https://discuss.elastic.co/u/Joe_Lawson)\ **Post date:** [June 10, 2016, 7:58pm UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/12 "2016-06-10T19:58:48Z")
Yeah you need to run 0.9 for now. ---
**Author:** ![upendra](https://avatars.discourse-cdn.com/v4/letter/u/e68b1a/32.png) [@upendra](https://discuss.elastic.co/u/upendra)\ **Post date:** [June 13, 2016, 9:26am UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/13 "2016-06-13T09:26:50Z")
Hi Joe, We have downgraded kafka to 0.9 but are still facing the same deserialization error (as shown above) while consuming with Kafka-Avro-Consumer. Looks like the Avro-Consumer is not picking up the right schema. Can you help? Thanks! ---
**Author:** ![Radek\_Tomsej](https://avatars.discourse-cdn.com/v4/letter/r/8baadc/32.png) [@Radek\_Tomsej](https://discuss.elastic.co/u/Radek_Tomsej)\ **Post date:** [July 6, 2016, 12:28pm UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/14 "2016-07-06T12:28:15Z")
There are two problems: 1. [Schema Registry](http://docs.confluent.io/3.0.0/schema-registry/docs/index.html) API returns the Avro schema in `schema` object. You expect that the Avro schema is not inside of nested object. You bypassed this problem with defining schema with a file from local disk (it is not ideal, but it works :). 2. Every message inside of Confluent Platform should have following structure: `< magic byte > < schema id (4 bytes) > < Avro blob >` Currently the codec is not prefixing the Avro blob with magic byte nor schema id. So the kafka-avro-console-consumer expects that inside first 5 bytes there is a magic byte and schema id so it cannot pick up the right schema. The only solution for this is to write a new codec with prefixing Avro blob. Should not be hard, but I am not a Ruby developer ☹ ---
**Author:** ![upendra](https://avatars.discourse-cdn.com/v4/letter/u/e68b1a/32.png) [@upendra](https://discuss.elastic.co/u/upendra)\ **Post date:** [July 11, 2016, 3:05pm UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/15 "2016-07-11T15:05:44Z")
HI Joe & Team, Any idea of a codec which prefixes Avro blob? Is there anything being worked on? Thanks, Upendra ---
**Author:** ![bruce.szalwinski](https://avatars.discourse-cdn.com/v4/letter/b/7feea3/32.png) [@bruce.szalwinski](https://discuss.elastic.co/u/bruce.szalwinski)\ **Post date:** [October 20, 2016, 3:58pm UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/16 "2016-10-20T15:58:29Z")
I'm working on a pull request against the logstash-codec-avro that will add support to strip the magic byte and schema id when the strip\_header config is set to true, [https://github.com/logstash-plugins/logstash-codec-avro/pull/20](https://github.com/logstash-plugins/logstash-codec-avro/pull/20) ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [July 6, 2017, 4:33am UTC](https://discuss.elastic.co/t/not-able-to-read-kafka-avro-schema-messages/24979/17 "2017-07-06T04:33:24Z")