title: How to configure Debezium Mysql connector to produce primitive key instead of struct or json object?
body: <p>I am using Debezium to detect changes in MySql source tables.   How can I produce Kafka messages such that the key is a numeric (<code>Long</code>) value instead of a Json object?</p>
  
  <p>What I am getting:</p>
  
  <pre><code>key: {"foo_id": 123} 
  value: {"foo_id": 123, "bar": "blahblah", "baz": "meh......"}
  </code></pre>
  
  <p>What I want:</p>
  
  <pre><code>key: 123
  value: {"foo_id": 123, "bar": "blahblah", "baz": "meh......"}
  </code></pre>
  
  <p>My FOO table looks like this:</p>
  
  <pre><code>foo_id: INT
  bar: VARCHAR 
  baz: VARCHAR
  </code></pre>
  
  <p>Note that I am not using avro and I've experimented with several combinations of below (w/ and w/o key transformers) but failed to get the <code>Long</code> key. </p>
  
  <pre><code>"transforms": "unwrap,insertKey,extractKey",
  "transforms.unwrap.type":"io.debezium.transforms.UnwrapFromEnvelope",
  "transforms.unwrap.drop.tombstones":"false",
  "transforms.insertKey.type":"org.apache.kafka.connect.transforms.ValueToKey",
  "transforms.insertKey.fields":"foo_id",
  "transforms.extractKey.type":"org.apache.kafka.connect.transforms.ExtractField$Key",
  "transforms.extractKey.field":"foo_id",        
  "key.converter" : "org.apache.kafka.connect.converters.LongConverter",
  "key.converter.schemas.enable": "false", 
  "value.converter" : "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "false"
  </code></pre>
  
  <p>I am not sure ValueToKey or ExtractField works for (MySQL) Source but I am getting below NPE.</p>
  
  <pre><code>Caused by: java.lang.NullPointerException
          at org.apache.kafka.connect.transforms.ValueToKey.applyWithSchema(ValueToKey.java:85)
          at org.apache.kafka.connect.transforms.ValueToKey.apply(ValueToKey.java:65)
          at org.apache.kafka.connect.runtime.TransformationChain.lambda$apply$0(TransformationChain.java:44)
          at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)
          at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162) 
  </code></pre>