我试图添加密钥使用smt函数使用它作为es文件的文件id，但它不工作。我正在使用汇合es连接器。下面的配置文件：connect-standalone.properties

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

连接器配置：

# 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

我向Kafka发出以下信息：

{"recordId":"999","activity":"test","description":"test Cont"}

在接收器连接器中获取此错误：

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)

请建议！我已经试着修了一段时间了。非常感谢您的帮助…提前谢谢