I am trying to use jdbc connector to send records from Kafka 0.9 to DB. I
am using jsonConverter to convert the records. My connector is failing when
its checking the Schema I am using. Please let me know what is the issue
with my json schema.
Configuration used:
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
# Converter-specific settings can be passed in by prefixing the Converter's
setting with the converter we want to apply
# it to
key.converter.schemas.enable=true
value.converter.schemas.enable=true
Record that has been sent to the topic -
{"schema":{"type":"struct","fields":[{"name":"error_code",
"type":"string","optional":"false"},{"name":"error_time","
type":"string","optional":"false"},{"name":"error_msg","
type":"string","optional":"false"},{"name":"source","
type":"string","optional":"false"},{"name":"criticality",
"type":"string","optional":"false"}]},"payload":{"error_
code":"RAW104","error_time":"09/15/2016@18:00:32","error_msg":"Not
accepting","source":"APPLICATION","criticality":"WARN"}}
Error I am seeing:
[2016-09-15 18:01:07,513] ERROR Thread WorkerSinkTask-jdbc-sink-test-0
exiting with uncaught exception: (org.apache.kafka.connect.
util.ShutdownableThread:84)
*org.apache.kafka.connect.errors.DataException: Struct schema's field name
not specified properly*
at org.apache.kafka.connect.json.JsonConverter.asConnectSchema(
JsonConverter.java:493)
at org.apache.kafka.connect.json.JsonConverter.jsonToConnect(
JsonConverter.java:344)
at org.apache.kafka.connect.json.JsonConverter.toConnectData(
JsonConverter.java:334)
at org.apache.kafka.connect.runtime.WorkerSinkTask.
convertMessages(WorkerSinkTask.java:266)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(
WorkerSinkTask.java:175)
at org.apache.kafka.connect.runtime.WorkerSinkTaskThread.
iteration(WorkerSinkTaskThread.java:90)
at org.apache.kafka.connect.runtime.WorkerSinkTaskThread.
execute(WorkerSinkTaskThread.java:58)
at org.apache.kafka.connect.util.ShutdownableThread.run(
ShutdownableThread.java:82)
Exception in thread "WorkerSinkTask-jdbc-sink-test-0"
*org.apache.kafka.connect.errors.DataException:
Struct schema's field name not specified properly*
at org.apache.kafka.connect.json.JsonConverter.asConnectSchema(
JsonConverter.java:493)
at org.apache.kafka.connect.json.JsonConverter.jsonToConnect(
JsonConverter.java:344)
at org.apache.kafka.connect.json.JsonConverter.toConnectData(
JsonConverter.java:334)
at org.apache.kafka.connect.runtime.WorkerSinkTask.
convertMessages(WorkerSinkTask.java:266)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(
WorkerSinkTask.java:175)
at org.apache.kafka.connect.runtime.WorkerSinkTaskThread.
iteration(WorkerSinkTaskThread.java:90)
at org.apache.kafka.connect.runtime.WorkerSinkTaskThread.
execute(WorkerSinkTaskThread.java:58)
at org.apache.kafka.connect.util.ShutdownableThread.run(
ShutdownableThread.java:82)
Thanks,
Sri
