Failed to deserialize data for topic

Issue fixed.

Basically kafka message was successfully persisted to the topic but when my JDBC sink connector was trying to parse it and copy to our MySQL DB it wasn't able to connect to the schema registry URL.

Previous connector config:

{ "name": "zuum-sink-positions", "key.converter.schema.registry.url": "http://localhost:8081", "value.converter.schema.registry.url": "http://localhost:8081", "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector", "key.converter": "org.apache.kafka.connect.storage.StringConverter", "value.converter": "io.confluent.connect.avro.AvroConverter", "key.converter.schemas.enable":"false", "value.converter.schemas.enable":"true", "config.action.reload": "restart", "errors.log.enable": "true", "errors.log.include.messages": "true", "print.key": "true", "errors.tolerance": "all", "topics": "zuum-positions", "connection.url": "jdbc:mysql://ip:3306/zuum_tracking", "connection.user": "user", "connection.password": "password", "auto.create": "true" } 

Updated schema registry url with correct host:

{ "name": "zuum-sink-positions", "key.converter.schema.registry.url": "http://schema-registry:8081", "value.converter.schema.registry.url": "http://schema-registry:8081", "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector", "key.converter": "org.apache.kafka.connect.storage.StringConverter", "value.converter": "io.confluent.connect.avro.AvroConverter", "key.converter.schemas.enable":"false", "value.converter.schemas.enable":"true", "config.action.reload": "restart", "errors.log.enable": "true", "errors.log.include.messages": "true", "print.key": "true", "errors.tolerance": "all", "topics": "zuum-positions", "connection.url": "jdbc:mysql://ip:3306/zuum_tracking", "connection.user": "user", "connection.password": "password", "auto.create": "true" } 

Assuming the issue was the key of the message, the alternative answer would be to change the Connect container env vars to change

CONNECT_KEY_CONVERTER=io.confluent.connect.avro.AvroConverter CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL=http://schema-registry:8081 

Otherwise, if the issue was only the value, then you only need to specify the converter settings within the JSON if you want to override the default values set in the Compose / connect-distributed.properties file. In other words, you could have removed the localhost values completely.

This error is not re-tried by Confluent Connect and if your tasks end up failing due to this, they could be possibly removed from the cluster. This error can also show up in a Production cluster running Confluent 5.5.x or earlier, I had it last night see below -

Graceful stop of task failed (Executing stage VALUE_CONVERTER) [2022-01-09 01:11:27,890] ERROR Graceful stop of task xyz_s3-sink209_2222_MESSAGE-0 failed. (org.apache.kafka.connect.runtime.Worker:746) [2022-01-09 01:11:31,589] ERROR Error encountered in task xyz_s3-sink209_2222_MESSAGE-0. Executing stage 'VALUE_CONVERTER' with class 'io.confluent.connect.avro.AvroConverter', where consumed record is {topic='b0213_BC_MESSAGE', partition=0, offset=6007846, timestamp=1641664763873, timestampType=CreateTime}. (org.apache.kafka.connect.runtime.errors.LogReporter:62) [2022-01-09 01:11:31,591] ERROR WorkerSinkTask{id=xyz_s3-sink209_2222_MESSAGE-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:186) [2022-01-09 01:11:31,594] ERROR WorkerSinkTask{id=xyz_s3-sink209_2222_MESSAGE-0} Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:187) 

And the Complete error could look like this below -

{"id":0,"state":"FAILED","worker_id":"xyz.co.uk:8083","trace":"org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:178)\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:491)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:468)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:324)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:228)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:200)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:184)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:234)\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:1149)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\n\tat java.lang.Thread.run(Thread.java:748)\nCaused by: org.apache.kafka.connect.errors.DataException: Failed to deserialize data for topic xyz_2222_MESSAGE to Avro: \n\tat io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:118)\n\tat org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:87)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$1(WorkerSinkTask.java:491)\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)\n\t... 13 more\nCaused by: org.apache.kafka.common.errors.SerializationException: Error retrieving Avro value schema for id 32818\nCaused by: java.net.SocketTimeoutException: Read timed out\n\tat java.net.SocketInputStream.socketRead0(Native Method)\n\tat java.net.SocketInputStream.socketRead(SocketInputStream.java:116)\n\tat java.net.SocketInputStream.read(SocketInputStream.java:171)\n\tat java.net.SocketInputStream.read(SocketInputStream.java:141)\n\tat java.io.BufferedInputStream.fill(BufferedInputStream.java:246)\n\tat java.io.BufferedInputStream.read1(BufferedInputStream.java:286)\n\tat java.io.BufferedInputStream.read(BufferedInputStream.java:345)\n\tat sun.net.www.http.HttpClient.parseHTTPHeader(HttpClient.java:735)\n\tat sun.net.www.http.HttpClient.parseHTTP(HttpClient.java:678)\n\tat sun.net.www.protocol.http.HttpURLConnection.getInputStream0(HttpURLConnection.java:1587)\n\tat sun.net.www.protocol.http.HttpURLConnection.getInputStream(HttpURLConnection.java:1492)\n\tat java.net.HttpURLConnection.getResponseCode(HttpURLConnection.java:480)\n\tat io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:272)\n\tat io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:351)\n\tat io.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:659)\n\tat io.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:641)\n\tat io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaByIdFromRegistry(CachedSchemaRegistryClient.java:217)\n\tat io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaBySubjectAndId(CachedSchemaRegistryClient.java:291)\n\tat io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaById(CachedSchemaRegistryClient.java:276)\n\tat io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer$DeserializationContext.schemaFromRegistry(AbstractKafkaAvroDeserializer.java:273)\n\tat io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer$DeserializationContext.getSubject(AbstractKafkaAvroDeserializer.java:297)\n\tat io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer$DeserializationContext.schemaForDeserialize(AbstractKafkaAvroDeserializer.java:286)\n\tat io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserializeWithSchemaAndVersion(AbstractKafkaAvroDeserializer.java:151)\n\tat io.confluent.connect.avro.AvroConverter$Deserializer.deserialize(AvroConverter.java:162)\n\tat io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:101)\n\tat org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:87)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$1(WorkerSinkTask.java:491)\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:491)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:468)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:324)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:228)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:200)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:184)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:234)\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:1149)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\n\tat java.lang.Thread.run(Thread.java:748)\n"} 

If Schema Registry instances are not running on all nodes within a load balanced cluster, you could see this error.

Any Connect tasks trying to deserialize data from Kafka would fail until Schema Registry is healthy and accessible.

Once it's verified that Kafka brokers and Schema Registry are back up and healthy, the Connect tasks could be unloaded and re-started possibly on a non-balancing connect node in the cluster to solve this problem.