'Unable to parse a topic from Kafka Connect AVRO connector
I am using a Neo4j sink connector to read data from a Kafka topic and dump it to a Neo4j database. The message/data available in Kafka is of AVRO format , hence I am trying to use an AVRO converter to parse the data by providing schema registry details. But while consuming the message I am seeing a DataError exception.
Below is my configuration to create the connector.
{
"topics": "mytopic",
"connector.class": "streams.kafka.connect.sink.Neo4jSinkConnector",
"tasks.max":"1",
"key.converter.schemas.enable":"true",
"values.converter.schemas.enable":"true",
"errors.retry.timeout": "-1",
"errors.retry.delay.max.ms": "1000",
"errors.tolerance": "none",
"errors.deadletterqueue.topic.name": "deadletter-topic",
"errors.deadletterqueue.topic.replication.factor":1,
"errors.deadletterqueue.context.headers.enable":true,
"key.converter":"org.apache.kafka.connect.storage.StringConverter",
"key.converter.enhanced.avro.schema.support":true,
"value.converter.enhanced.avro.schema.support":true,
"value.converter":"io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url":"https://schema-url/",
"value.converter.basic.auth.credentials.source":"USER_INFO",
"value.converter.basic.auth.user.info":"user:pass",
"errors.log.enable": true,
"schema.ignore":"false",
"errors.log.include.messages": true,
"neo4j.server.uri": "neo4j://my-ip:7687/neo4j",
"neo4j.authentication.basic.username": "neo4j",
"neo4j.authentication.basic.password": "neo4j",
"neo4j.encryption.enabled": false,
"neo4j.topic.cypher.mytopic": "MERGE (p:Loc_Con{name: event.geography.name})"
}
Here is the excpetion I am getting.
ErrorData(originalTopic=mytopic, timestamp=1652188554497, partition=0, offset=2140111, exception=org.apache.kafka.connect.errors.DataException: Exception thrown while processing field 'geography', key=9662840 , value=Struct{geography=Struct{geoId=43333,geoType=Business Defined Area,name=Norarea,status=Active,validFrom=Sat Apr 09 00:00:00 GMT 2012,validTo=Fri Dec 31 00:00:00 GMT 9999, executingClass=class streams.kafka.connect.sink.Neo4jSinkTask)
I wanted to know what is going wrong here , also I tried with String and JSON Converter as well ,but there also the parsing is failing. So is there any option to parse the data ?
Sources
This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.
Source: Stack Overflow
Solution | Source |
---|