Aiven-Open / Aiven-Open/opensearch-connector-for-apache-kafka
Failed to deserialize data for topic [topic] to Avro
- Dominant language
- Java
- Stars
- 95
- Forks
- 61
- Avg merge
- 11h 52m
- Merged PRs (30d)
- 7
Description
Hi,
The plugin, from GitHub's main branch, throws an error when consuming a topic with simple json as well as a topic with avro messages.
I'm using the latest versions of confluentinc Broker, Zookeeper, Connect, and opensearchproject OpenSearch.
I wonder if this is related to the following variable when starting Connect:
` CONNECT_KEY_CONVERTER: io.confluent.connect.avro.AvroConverter`
My config:
```
{
"name": "opensearch-sink",
"config": {
"connector.class": "io.aiven.kafka.connect.opensearch.OpensearchSinkConnector",
"tasks.max": "1",
"topics": "some-existing-topic",
"key.ignore": "true",
"connection.url": "http://opensearch-node1:9200",
"type.name": "kafka-connect"
}
}
```
Connect stack trace
```
[2022-12-09 21:02:36,297] ERROR WorkerSinkTask{id=opensearch-sink-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask)
2022-12-09T21:02:36.298045183Z org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
2022-12-09T21:02:36.298063297Z at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:223)
2022-12-09T21:02:36.298067853Z at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:149)
2022-12-09T21:02:36.298071164Z at org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:513)
2022-12-09T21:02:36.298074122Z at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:493)
2022-12-09T21:02:36.298077269Z at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:332)
2022-12-09T21:02:36.298080134Z at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:234)
2022-12-09T21:02:36.298082970Z at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:203)
2022-12-09T21:02:36.298086088Z at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:188)
2022-12-09T21:02:36.298089054Z at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:243)
2022-12-09T21:02:36.298092333Z at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
2022-12-09T21:02:36.298095317Z at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
2022-12-09T21:02:36.298098179Z at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
2022-12-09T21:02:36.298101007Z at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
2022-12-09T21:02:36.298103968Z at java.base/java.lang.Thread.run(Thread.java:829)
2022-12-09T21:02:36.298107141Z Caused by: org.apache.kafka.connect.errors.DataException: Failed to deserialize data for topic role-received to Avro:
2022-12-09T21:02:36.298110427Z at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:124)
2022-12-09T21:02:36.298113334Z at org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:88)
2022-12-09T21:02:36.298116211Z at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$3(WorkerSinkTask.java:513)
2022-12-09T21:02:36.298122519Z at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:173)
2022-12-09T21:02:36.298125917Z at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:207)
2022-12-09T21:02:36.298129072Z ... 13 more
2022-12-09T21:02:36.298131874Z Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!
2022-12-09T21:02:36.298134802Z at io.confluent.kafka.serializers.AbstractKafkaSchemaSerDe.getByteBuffer(AbstractKafkaSchemaSerDe.java:250)
2022-12-09T21:02:36.298174390Z at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer$DeserializationContext.(AbstractKafkaAvroDeserializer.java:334)
2022-12-09T21:02:36.298181482Z at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserializeWithSchemaAndVersion(AbstractKafkaAvroDeserializer.java:202)
2022-12-09T21:02:36.298184797Z at io.confluent.connect.avro.AvroConverter$Deserializer.deserialize(AvroConverter.java:172)
2022-12-09T21:02:36.298187722Z at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:107)
2022-12-09T21:02:36.298190604Z ... 17 more
```
Contributor guide
Research direction
Start with the reported Connect configuration and the AvroConverter and WorkerSinkTask paths shown in the stack trace; reproduce the failure against the JSON and Avro topics described. Compare the converter's expectations with the data on each topic, then document or implement the confirmed resolution and verify that consumption succeeds.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, kafka
- Domain
- distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100