title: How to use the AVRO serializer with Schema Registry from a Kafka Connect SourceTask
body: <p>I have set up the Confluence Data Platform and started to develop a SourceConnector and in the corresponding SourceTask.poll() method I do the following (pseudo Java code below):</p>
  
  <pre><code>    public List&lt;SourceRecord&gt; poll() throws InterruptedException {
  
  ....
  
      Envelope envelope = new Envelope();
      ByteArrayOutputStream out = new ByteArrayOutputStream();
      Encoder enc = EncoderFactory.get().binaryEncoder(out, null);
      DatumWriter&lt;Envelope&gt; dw = new ReflectDatumWriter&lt;Envelope&gt;(Envelope.class);
      dw.write((Envelope)envelope, enc);
      enc.flush();
      out.close();
      Map&lt;String, String&gt; sourcePartition = new HashMap&lt;String, String&gt;();
      sourcePartition.put("stream", streamName);
      Map&lt;String, Integer&gt; sourceOffset = new HashMap&lt;String, Integer&gt;();
      sourceOffset.put("position", Integer.parseInt(envelope.getTimestamp()));
      records.add(new SourceRecord(sourcePartition, sourceOffset, topic, org.apache.kafka.connect.data.Schema.BYTES_SCHEMA, envelope));
  
  ....
  </code></pre>
  
  <p>I'd like to use the Schema Registry so that the object being serialized is tagged with a schema id from the registry, serialized and then published to the Kafka-topic through the poll() function. If the schema for an arbitrary object doesn't reside in the registry I want it to be registered and the corresponding generated id returned to the serializer process so it becomes part of the serialized object making it deserializable.</p>
  
  <p>What do I need to do in the code above to make this happen?</p>