即使json数据包含模式和有效负载字段，kafka connect hdfs接收器连接器也会失败

Question

我正在尝试kafka connect hdfs接收器连接器，以便将json数据从kafka移动到hdfs。

即使在kafka中的json数据具有模式和有效负载时，kafka connect任务也会因错误而失败

org.apache.kafka.connect.errors.DataException: JsonConverter with schemas.enable requires \"schema\" and \"payload\" fields and may not contain additional fields.

Kafka的数据：

{"schema": {"type": "struct","fields": [{"type": "string","optional": false,"field": "Name"}, {"type": "string","optional": false,"field": "company"}],"optional": false,"name": "Person"},"payload": {"Name": "deepak","company": "BT"}}
{"schema": {"type": "struct","fields": [{"type": "string","optional": false,"field": "Name"}, {"type": "string","optional": false,"field": "company"}],"optional": false,"name": "Person"},"payload": {"Name": "sufi","company": "BT"}}
{"schema": {"type": "struct","fields": [{"type": "string","optional": false,"field": "Name"}, {"type": "string","optional": false,"field": "company"}],"optional": false,"name": "Person"},"payload": {"Name": "vikas","company": "BT"}}

curl -X POST -H "Content-Type: application/json" --data '{"name": "connect-cluster-15may-308pm", "config": {"connector.class":"io.confluent.connect.hdfs.HdfsSinkConnector", "tasks.max":"2", "hdfs.url": "hdfs://localhost:9000","flush.size": "1","topics":"test_hdfs_json_schema_payload_1","topics.dir":"/deepak/kafka-connect/trial1"}}' http://localhost:8083/connectors

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true

错误消息：

http://localhost:8083/connectors/connect-cluster-15may-308pm/tasks/0/status

{
    "state": "FAILED",
    "trace": "org.apache.kafka.connect.errors.DataException: JsonConverter with schemas.enable requires \"schema\" and \"payload\" fields and may not contain additional fields. If you are trying to deserialize plain JSON data, set schemas.enable=false in your converter configuration.\n\tat org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:309)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:400)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:249)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:179)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:148)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:139)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:182)\n\tat java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)\n\tat java.util.concurrent.FutureTask.run(FutureTask.java:266)\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)\n\tat java.lang.Thread.run(Thread.java:745)\n",
    "id": 0,
    "worker_id": "127.0.0.1:8083"
}

Answer 1

你用的是什么版本的Kafka连接？当从stacktrace中确定错误的来源时，了解这一点很有帮助。

我认为发生的情况是，你在值中有数据，但在键中没有。由于您将key.converter和value.converter都设置为JSONConverter并且使用schemas.enable=true，因此希望看到包含schema和payload的信封格式。但是，我猜您的密钥都是null。

这与https://issues.apache.org/jira/browse/kafka-3832类似，其中jsonConverter从不生成真null值。相反，它总是生成包含预期的可选模式+null负载的信封。在这种情况下，从Kafka转换到Connect的数据API是不起作用的，因为它在密钥中期望相同的信封格式。

您可以通过向控制台使用者命令添加--property print.key=true来验证是否存在问题。如果它打印出null键，问题是JsonConverter无法解码它们。

一个简单的解决方法是，只对不关心空值的键使用其他转换器--键中没有任何数据。Kafka Connect附带的一个工具是org.apache.Kafka.Connect.storage.StringConverter。