Kafka-ES-Sink : ConnectException: Key 用作文档 ID，不能为空

Question

我正在尝试使用 SMT 函数添加密钥以将其用作 ES 文档的文档 ID，但它不起作用。我正在使用 confluent es connector。配置文件如下：连接独立属性

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 

我正在向卡夫卡发送以下消息：

{"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)

请指教！一段时间以来一直在尝试修复它。任何帮助都非常感谢...提前致谢

Answer 1

我通过更改以下属性解决了这个问题。

key.ignore=true