title: kafka FileStreamSourceConnector write an avro file to topic with key field
body: <p>I want to use kafka <code>FileStreamSourceConnector</code> to write a local avro file into a topic.</p>
  <p>My connector config looks like this:</p>
  <pre class="lang-sh prettyprint-override"><code>curl -i -X PUT -H  &quot;Content-Type:application/json&quot; http://localhost:8083/connectors/file_source_connector/config \
              -d '{
              &quot;connector.class&quot;: &quot;org.apache.kafka.connect.file.FileStreamSourceConnector&quot;,
              &quot;value.converter.schema.registry.url&quot;: &quot;http://schema-registry:8081&quot;,
              &quot;topic&quot;: &quot;my_topic&quot;,
              &quot;file&quot;: &quot;/data/log.avsc&quot;,
              &quot;format.include.keys&quot;: &quot;true&quot;,
              &quot;source.auto.offset.reset&quot;: &quot;earliest&quot;,
              &quot;tasks.max&quot;: &quot;1&quot;,
              &quot;value.converter.schemas.enable&quot;: &quot;true&quot;,
              &quot;value.converter&quot;: &quot;io.confluent.connect.avro.AvroConverter&quot;,
              &quot;key.converter&quot;: &quot;org.apache.kafka.connect.storage.StringConverter&quot;
            }'
  </code></pre>
  <p>Then when I print out the topic, the key fields are <code>null</code>.</p>
  <p><strong>Updated on 2021-03-29</strong>:</p>
  <p>After watching this video <a href="https://www.youtube.com/watch?v=gSaCtaHt1k4&amp;list=PL5T99fPsK7pq7LiaaL-S6b7wQqzxyjgya&amp;index=2" rel="nofollow noreferrer">Twelve Days of SMT  - Day 2: ValueToKey and ExtractField</a> from Robin,
  I applied SMT to my connector config:</p>
  <pre class="lang-sh prettyprint-override"><code>curl -i -X PUT -H  &quot;Content-Type:application/json&quot; http://localhost:8083/connectors/file_source_connector_02/config \
              -d '{
              &quot;connector.class&quot;: &quot;org.apache.kafka.connect.file.FileStreamSourceConnector&quot;,
              &quot;value.converter.schema.registry.url&quot;: &quot;http://schema-registry:8081&quot;,
              &quot;topic&quot;: &quot;my_topic&quot;,
              &quot;file&quot;: &quot;/data/log.avsc&quot;,
              &quot;tasks.max&quot;: &quot;1&quot;,
              &quot;value.converter&quot;: &quot;io.confluent.connect.avro.AvroConverter&quot;,
              &quot;key.converter&quot;: &quot;org.apache.kafka.connect.storage.StringConverter&quot;,
              &quot;transforms&quot;: &quot;ValueToKey, ExtractField&quot;,
              &quot;transforms.ValueToKey.type&quot;:&quot;org.apache.kafka.connect.transforms.ValueToKey&quot;,
              &quot;transforms.ValueToKey.fields&quot;:&quot;id&quot;,
              &quot;transforms.ExtractField.type&quot;:&quot;org.apache.kafka.connect.transforms.ExtractField$Key&quot;,
              &quot;transforms.ExtractField.field&quot;:&quot;id&quot;
            }'
  </code></pre>
  <p>However, the connector is failed:</p>
  <pre class="lang-sh prettyprint-override"><code>Caused by: org.apache.kafka.connect.errors.DataException: Only Struct objects supported for [copying fields from value to key], found: java.lang.String
  </code></pre>