Kafka-connect-jdbc: JDBC Sink: JsonConverter with schemas.enable requires "schema" and "payload" fields and may not contain additional fields.

[manikanthkoora]

Even though data has schema and payload, I am getting this issue.Could you help me to fix this issue. Thanks.

ERROR WorkerSinkTask{id=topics1-bulk-sink-9} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:172)
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.
        at org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:308)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:453)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:287)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:198)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:166)
        at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
        at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)

[rhauch]

Can you provide an examples of a message key and message value? This error message happens when the structure contains fields other than schema and payload, which is the envelope structure used by the JsonConverter with schemas.enable=true (the default).

[manikanthkoora]

Hi rhauch,

Thanks for responding to my question. My messages contain only schema and payload. Here I am attaching the example message.

{"schema":{"type":"struct","fields":[{"type":"string","optional":false,"field":"EQUIP"},{"type":"int64","optional":true,"field":"NUM_EQUIP"},{"type":"string","optional":false,"field":"UNID"}],"optional":false,"name":"ACT_EQUIP"},"payload":{"EQUIP":"7KTUA28278","NUM_EQUIP":1,"UNID":"WG33609"}}

I am using the following properties.

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

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

Please let me know if I am doing something wrong. Thanks.

[rhauch]

What version of Connect are you using? We recently improved the error message to say whether the conversion failed on the message key or value. Since your error message doesn't include that, I wonder if your message keys also are JSON with the schema and payload.

[manikanthkoora]

I am using kafka-connect-jdbc-5.1.0.jar in Kafka connect. I am trying to read oracle db tables and creating topics on Kafka cluster. I am using jbdc source connector and its working fine. I am facing this issue when running jdbc sink connector. I don't think, I have message keys assigned to messages.

Oracle to Kafka Topics is done by Kafka Connect JDBC source connector.
Kafta Topics to another oracle db is done by Kafka Connect JDBC sink connector.

Could you please let me know, how to fix this issue. Thanks.

[rhauch]

Based upon your stack trace, it looks like you're using Kafka 1.0.0 (or Confluent Platform 4.0.0); later versions would have a stack trace line referencing line 338 or 337:

at org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:338)

Anyway, look at your JDBC sink connector configuration. In particular, you should either have the following lines in the connector configuration:

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

or these four properties should be completely absent. If they are set to anything else, then that is likely causing the problem.

[rhauch]

Another possibility is that you have some records that have a normal JSON format and others that are serialized differently. If this is the case, the sink connector would work for a bit and then fail.

[fzucxl]

@manikanthkoora，
I am run into the same problem yesterday and I resolve it by delete the topic and create again. The problem is you topic is not empty and the message in your topic can not be convert by JsonConverter.

[dhoppenb]

I recently had the same problem on an older environment but couldn't reproduce in one with more recent versions.
After upgrading the older one from [confluent-kafka-2.11-2.1.0-1, confluent-kafka-connect-jdbc-5.1.0-1] to [confluent-kafka-connect-jdbc-5.1.2-1, confluent-kafka-2.11-2.1.1cp1-1] the problem was gone.

[yogi12121993]

I am using kafka-connect-jdbc-5.1.0.jar in Kafka connect. I am trying to read oracle db tables and creating topics on Kafka cluster. I am using jbdc source connector and its working fine. I am facing this issue when running jdbc sink connector. I don't think, I have message keys assigned to messages.

Oracle to Kafka Topics is done by Kafka Connect JDBC source connector.
Kafta Topics to another oracle db is done by Kafka Connect JDBC sink connector.

Could you please let me know, how to fix this issue. Thanks.

Have you got any solution for this?
I mean for having same topic as created by source connector for sink connector.
Beacuse when running sink connector it will again create topic then both will not be in sync. right?

[filipius]

I lost a couple of hours around a problem like this. The sink reads a key-value pair from the topic. In my case the crash occurred because the sink was expecting a full JSON schema in the key, which simply was not there (a simple "1" string was). In my case I solved the problem by either outputting a null in place of the "1" (as the null will work with the key JSON unmarshaller) or more cleanly by adding this key-value pairs to the property file:

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

That is it.

To help me figure out the problem I ran the following command to read the messages in the topic (note the last option to print the key):

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test22 --from-beginning --property print.key=true

and I was getting this:

1 {"schema":{"type":"struct","fields":[{"type":"double","optional":false,"field":"revenue"},{"type":"double","optional":false,"field":"expenses"},{"type":"double","optional":false,"field":"profit"}],"optional":false,"name":"total data"},"payload":{"revenue":3542475.0, "expenses":2451680.0,"profit":1090795.0}}

Note the String in the key, a tab character and then the value with a {"schema": ... , "payload": ...} JSON schema. With the above options this key will work fine and the JsonConverter will indeed find what it expects for the value.

[ashwani29]

I got it working by setting schemas.enable for key and value to false as it suggests in the error message in both standalone-server and sink properties file.