KafkaConnect elasticSearch Document Id Creation

Question

我当前正在使用以下连接器配置，但出现异常"密钥用作文档 ID，不能为空">

{
"name" :"hello7",
"config" : {
"name": "hello7",
"connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"tasks.max": "1",
"topics": "hello7",
"connection.url":"http://127.0.0.1:8080/",
"type.name":"aggregator",
"schema.ignore": "true",
"topic.schema.ignore": "true",
"topic.key.ignore": "false",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false", 
"key.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable": "false",
"key.ignore":"false",
"transforms": "extractKey",
"transforms.InsertKey.type":"org.apache.kafka.connect.transforms.ValueToKey",
"transforms.InsertKey.fields":"customerId",
"transforms.extractKey.type":"org.apache.kafka.connect.transforms.ExtractField$Key",
"transforms.extractKey.field":"customerId",
"errors.log.enable":true,
"errors.log.include.messages":true
}
}

我正在向主题发送以下消息

{
"customerId" : "i7y32o4823",
"customerName" : "JOE",
"address":"123 main street",
"employee" : "ABC Company",
"employeeAddress" : "178 Main Street"
}

我收到以下错误

2020-01-17 16：28：33,624] 错误 WorkerSinkTask{id=hello7-0} 抛出的任务 未捕获且不可恢复的异常。任务正在被杀死并将 在手动重新启动之前无法恢复。 (org.apache.kafka.connect.runtime.WorkerSinkTask( org.apache.kafka.connect.errors.ConnectException：密钥用作 文档 ID，不能为空。 at io.confluent.connect.elasticsearch.DataConverter.convertKey(DataConverter.java：79( at io.confluent.connect.elasticsearch.DataConverter.convertRecord(DataConverter.java：160( at io.confluent.connect.elasticsearch.ElasticsearchWriter.tryWriteRecord(ElasticsearchWriter.java：285( at io.confluent.connect.elasticsearch.ElasticsearchWriter.write(ElasticsearchWriter.java：270( at io.confluent.connect.elasticsearch.ElasticsearchSinkTask.put(ElasticsearchSinkTask.java：169(

Answer 1

您已经设置了"key.ignore":"false"，并且只提到了要发送到 Kafka 主题的值。

Kafka 记录同时具有键和值。如果未指定键，则该键将为 null。

Elasticsearch 接收器连接器不接受空键，如错误所示

连接异常：密钥用作文档 ID，不能为空

此外，您只提取转换中的密钥，从不使用InsertKey

"transforms": "extractKey",

---

您可以使用控制台的独立文件接收器连接器输出来调试连接器

name=local-console-sink
connector.class=org.apache.kafka.connect.file.FileStreamSinkConnector
tasks.max=1
topics=hello7