Major Description
If you configure KafkaConnect to parse JSON messages, and you send it a non-JSON message, the SerializationException message will bubble up to the top, and stop KafkaConnect. While I understand sending non-JSON to a JSON serializer is a bad idea, I think that a single malformed message stopping all of KafkaConnect is even worse.
The data exception is thrown here: https://github.com/apache/kafka/blob/trunk/connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java#L305
From the call here: https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java#L476
This bubbles all the way up to the top, and KafkaConnect simply stops with the message: ERROR WorkerSinkTask{id=elasticsearch-sink-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:172)
Thoughts on adding a try/catch around the for loop in WorkerSinkTask's convertMessages so messages that don't properly parse are logged, but simply ignored? This way KafkaConnect can keep working even when it encounters a message it cannot decode?