title: Kafka-ES-Sink : ConnectException: Key is used as document id and can not be null
body: <p>I am trying to add key using SMT functions to use it as document id for ES document but it is not working. I am using confluent es connector. Config file below:
  <strong>connect-standalone.properties</strong></p>
  
  <pre><code>bootstrap.servers=localhost:9092
  key.converter=org.apache.kafka.connect.json.JsonConverter
  value.converter=org.apache.kafka.connect.json.JsonConverter
  key.converter.schemas.enable=false
  value.converter.schemas.enable=false
  internal.key.converter=org.apache.kafka.connect.json.JsonConverter
  internal.value.converter=org.apache.kafka.connect.json.JsonConverter
  internal.key.converter.schemas.enable=false
  internal.value.converter.schemas.enable=false
  offset.storage.file.filename=/tmp/connect.offsets
  offset.flush.interval.ms=10000
  </code></pre>
  
  <p><strong>Connector Config:</strong></p>
  
  <pre><code>#Connector name                                                                  
  name=logs-=false                                                                 
  #Connector class                                                                 
  connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector    
  tasks.max=1                                                                      
  topics=test                                                                  
  topic.index.map=test:activity                                     
  connection.url=http://localhost:9200                                             
  type.name=Activity                                                      
  #ignore key and schema                                                           
  key.ignore=false                                                                 
  schema.ignore=true                                                               
  transforms=InsertKey,ExtractId                                                   
  transforms.InsertKey.type=org.apache.kafka.connect.transforms.ValueToKey         
  transforms.InsertKey.fields=recordId                                          
  transforms.ExtractId.type=org.apache.kafka.connect.transforms.ExtractField$Key   
  transforms.ExtractId.field=recordId 
  </code></pre>
  
  <p>I am sending below message to kafka : </p>
  
  <pre><code>{"recordId":"999","activity":"test","description":"test Cont"}
  </code></pre>
  
  <p>Getting this error in sink connector:</p>
  
  <pre><code>org.apache.kafka.connect.errors.ConnectException: Key is used as document id and can not be null.
          at io.confluent.connect.elasticsearch.DataConverter.convertKey(DataConverter.java:56)
          at io.confluent.connect.elasticsearch.DataConverter.convertRecord(DataConverter.java:86)
          at io.confluent.connect.elasticsearch.ElasticsearchWriter.write(ElasticsearchWriter.java:210)
          at io.confluent.connect.elasticsearch.ElasticsearchSinkTask.put(ElasticsearchSinkTask.java:119)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:384)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:240)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:172)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:143)
          at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:140)
          at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:175)
          at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
          at java.util.concurrent.FutureTask.run(Unknown Source)
          at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
          at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
          at java.lang.Thread.run(Unknown Source)
  [2017-10-15 01:37:41,789] ERROR Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerSinkTask:405)
  [2017-10-15 01:37:41,789] INFO WorkerSinkTask{id=logs-=false-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSinkTask:262)
  [2017-10-15 01:37:41,804] ERROR Task logs-=false-0 threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:142)
  org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
          at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:406)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:240)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:172)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:143)
          at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:140)
          at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:175)
          at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
          at java.util.concurrent.FutureTask.run(Unknown Source)
          at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
          at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
          at java.lang.Thread.run(Unknown Source)
  </code></pre>
  
  <p>Please suggest! Have been trying to fix it for some time now. Any help is much appreciated...Thanks in advance</p>