[KAFKA-5996] JsonConverter generates "Mismatching schema" DataException

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.10.0.0, 0.10.0.1, 0.10.1.0, 0.10.1.1, 0.10.2.0, 0.10.2.1, 0.11.0.0
• Fix Version/s: 1.0.2, 1.1.0
• Component/s: connect
• Assignee: Unassigned
• Reporter: Yuqi Li
• Created: 29/Sep/17 19:27
• Updated: 14/Feb/18 16:52
• Resolved: 14/Feb/18 16:52

Description

Run connect-standalone.sh with original connect-standalone.properties and custom jdbc source connect configuration.

part of the jdbc source connect configuration:

connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
query=Select * from ......
poll.interval.ms=45000
transforms=MakeMap, InsertSourceId, InsertSourceEvent
transforms.MakeMap.type=org.apache.kafka.connect.transforms.HoistField$Value
transforms.MakeMap.field=emailRequest
transforms.InsertSourceId.type=org.apache.kafka.connect.transforms.InsertField$Value
transforms.InsertSourceId.static.field=id
transforms.InsertSourceId.static.value=1
transforms.InsertSourceEvent.type=org.apache.kafka.connect.transforms.InsertField$Value
transforms.InsertSourceEvent.static.field=event
transforms.InsertSourceEvent.static.value=REMINDER_EMAIL

When the source connect starts, the first iteration(executing the query) runs ok, but 45 seconds later, it throws following exception:
org.apache.kafka.connect.errors.DataException: Mismatching schema.
at org.apache.kafka.connect.json.JsonConverter.convertToJson(JsonConverter.java:647)
at org.apache.kafka.connect.json.JsonConverter.convertToJson(JsonConverter.java:650)
at org.apache.kafka.connect.json.JsonConverter.convertToJsonWithEnvelope(JsonConverter.java:537)
at org.apache.kafka.connect.json.JsonConverter.fromConnectData(JsonConverter.java:290)
at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:198)
at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:168)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:146)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:190)
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-09-29 13:59:18,361] ERROR Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:149)
The SoureTask is being killed due to the exception, and the source connector is dead.

Issue Links

• links to: GitHub Pull Request #4523