title: Kafka producer: send avro as array[byte] without schema
body: <p>I am trying to setup a simple kafka stack in local and I am at the point where I need to create a toy Producer. This: <a href="https://lombardo-chcg.github.io/tools/2017/09/29/kafka-avro-producer-in-scala.html" rel="nofollow noreferrer">https://lombardo-chcg.github.io/tools/2017/09/29/kafka-avro-producer-in-scala.html</a> (see below for the piece of code I'm interested in) is almost exactly what I want except:</p>
  <p>Here the producer sends a GenericData.Record object, so the whole schema is sent and it doesn't leverage the schema registry. I want to send an Array[Byte] with the first few bytes being the id of the schema and the following bytes being the data, without the schema (or so I think it is the optimal way to do it)</p>
  <p>The piece of code I am talking about:</p>
  <pre><code>import java.util.Properties
  
  import org.apache.avro.Schema.Parser
  import org.apache.avro.generic.GenericData
  import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
  import org.slf4j.LoggerFactory
  
  case class User(name: String, favoriteNumber: Int, favoriteColor: String)
  
  class AvroProducer {
    val logger = LoggerFactory.getLogger(getClass)
  
    val kafkaBootstrapServer = sys.env(&quot;KAFKA_BOOTSTRAP_SERVER&quot;)
    val schemaRegistryUrl = sys.env(&quot;SCHEMA_REGISTRY_URL&quot;)
  
    val props = new Properties()
    props.put(&quot;bootstrap.servers&quot;, kafkaBootstrapServer)
    props.put(&quot;schema.registry.url&quot;, schemaRegistryUrl)
    props.put(&quot;key.serializer&quot;, &quot;io.confluent.kafka.serializers.KafkaAvroSerializer&quot;)
    props.put(&quot;value.serializer&quot;, &quot;io.confluent.kafka.serializers.KafkaAvroSerializer&quot;)
    props.put(&quot;acks&quot;, &quot;1&quot;)
  
    val producer = new KafkaProducer[String, GenericData.Record](props)
    val schemaParser = new Parser
  
    val key = &quot;key1&quot;
    val valueSchemaJson =
    s&quot;&quot;&quot;
      {
        &quot;namespace&quot;: &quot;com.avro.junkie&quot;,
        &quot;type&quot;: &quot;record&quot;,
        &quot;name&quot;: &quot;User2&quot;,
        &quot;fields&quot;: [
          {&quot;name&quot;: &quot;name&quot;, &quot;type&quot;: &quot;string&quot;},
          {&quot;name&quot;: &quot;favoriteNumber&quot;,  &quot;type&quot;: &quot;int&quot;},
          {&quot;name&quot;: &quot;favoriteColor&quot;, &quot;type&quot;: &quot;string&quot;}
        ]
      }
    &quot;&quot;&quot;
    val valueSchemaAvro = schemaParser.parse(valueSchemaJson)
    val avroRecord = new GenericData.Record(valueSchemaAvro)
  
    val mary = new User(&quot;Mary&quot;, 840, &quot;Green&quot;)
    avroRecord.put(&quot;name&quot;, mary.name)
    avroRecord.put(&quot;favoriteNumber&quot;, mary.favoriteNumber)
    avroRecord.put(&quot;favoriteColor&quot;, mary.favoriteColor)
  
    def start = {
      try {
        val record = new ProducerRecord(&quot;users&quot;, key, avroRecord)
        val ack = producer.send(record).get()
        // grabbing the ack and logging for visibility
        logger.info(s&quot;${ack.toString} written to partition ${ack.partition.toString}&quot;)
      }
      catch {
        case e: Throwable =&gt; logger.error(e.getMessage, e)
      }
    }
  }
  </code></pre>
  <p>Problem(s):</p>
  <ul>
  <li>I don't know how to retrieve the id of the schema from schema-registry</li>
  <li>I don't know how to send only the data without the schema + the id as Array[Byte]</li>
  </ul>
  <p>I know how to write the whole avro to Array[Byte]:</p>
  <pre><code>    val writer = new SpecificDatumWriter[GenericData.Record](valueSchemaAvro)
      val out = new ByteArrayOutputStream
      val encoder = EncoderFactory.get.binaryEncoder(out, null)
      writer.write(avroRecord, encoder) // but here I am also writing the schema, right?
      encoder.flush
      out.close
      out.toByteArray
  </code></pre>
  <p>thanks so much</p>