'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