title: Unable to read avro messages using kafka-avro-console-consumer. SerializationException: Unknown magic byte
body: <p>I am writing a REST proxy like the confluent rest proxy. It takes a JSON payload, schema subject, and id, and then writes the JSON payload as an Avro object into the stream. When I use kafka-avro-console-consumer to read the message, I am getting "unknown magic byte" errors.</p>
  
  <p>Here is my kafka producer config:</p>
  
  <pre><code>        properties.put("client.id", LocalHostUtils.getLocalHostName(null));
  
          properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
          properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
  
          properties.put(AbstractKafkaAvroSerDeConfig.AUTO_REGISTER_SCHEMAS, false);
          properties.put(KafkaAvroSerializerConfig.VALUE_SUBJECT_NAME_STRATEGY, RecordNameStrategy.class);
  
          properties.put("schema.registry.url", configValuesManager.getString("dsp_kafka.schema_registry"));
  
          if (KafkaUtils.isKafkaEnabled()) {
              this.kafkaProducer = new KafkaProducer&lt;String, Object&gt;(properties);
          }
  </code></pre>
  
  <p>This is how the REST controller converts the incoming JSON to Avro</p>
  
  <pre><code>        Schema schema = null;
          try {
              schema = schemaRegistryClient.getBySubjectAndID(schemaSubject, schemaId);
          } catch (RestClientException e) {
              throw new IOExceptionWithCause(e);
          }
  
          log.debug(postContent);
          log.info("Subject/Version {}/{} -&gt; {}", schemaSubject, schemaId, schema);
          Object data = toAvro(schema, postContent);
  </code></pre>
  
  <p>This is the implementation of the <code>toAvro</code> method:</p>
  
  <pre><code>    Object toAvro(Schema schema, String jsonBody) throws IOException
      {
          DatumReader&lt;Object&gt; reader = new GenericDatumReader&lt;Object&gt;(schema);
          Object object = reader.read(
                  null, decoderFactory.jsonDecoder(schema, jsonBody));
  
          return object;
  
      }
  </code></pre>
  
  <p>This object is then passed to the schemaValidatingProducer that I configured with properties given above....</p>
  
  <pre><code>        this.kafkaSchemaValidatingProducer.publish(topic, 0, UUID.randomUUID().toString(), data);
  </code></pre>
  
  <p>This is the <code>publish</code> method on the <code>kafkaSchemaValidatingProducer</code></p>
  
  <pre><code>    public void publish(String topic, Integer partition, String key, Object data)
      {
          log.debug("publish topic={} key={} value={}", topic, key, data);
  
          if (!KafkaUtils.isKafkaEnabled()) {
              log.warn("Kafka is not enabled....");
              return;
          }
  
          ProducerRecord&lt;String, Object&gt; record = new ProducerRecord&lt;String, Object&gt;(topic, key, data);
  
  
          Future&lt;RecordMetadata&gt; metadataFuture = kafkaProducer.send(record, new Callback()
          {
              @Override
              public void onCompletion(RecordMetadata metadata, Exception exception)
              {
                  if (exception == null) {
                      log.info(metadata.toString());
                      return;
                  }
  
                  log.error("exception", exception);
  
              }
          });
  
          kafkaProducer.flush();
  
      }
  
  </code></pre>
  
  <p>this is how I am reading the topic</p>
  
  <pre><code>./bin/kafka-avro-console-consumer --bootstrap-server kafka-broker1:9021 --consumer.config client-ssl.properties --topic schema-validated-topic --property print.key=true --property print.value=true --value-deserializer io.confluent.kafka.serializers.KafkaAvroDeserializer --offset earliest --skip-message-on-error --partition 0 --property schema.registry.url http://schema-regisry
  </code></pre>
  
  <p>This results in....</p>
  
  <pre><code>[2019-08-26 16:30:36,351] ERROR Error processing message, skipping this message:  (kafka.tools.ConsoleConsumer$:76)
  org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
  Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!
  </code></pre>
  
  <p>Any idea why I am getting the "Bad magic number error" ? </p>