Kafka Connect Source connector problems with AVRO serialization #616
Comments
|
Hi @Romsick. Thank you for flagging it. Regarding nullable properties and the error you receive on null company property. Neo4j is a schema-less database. Schema is derived based on individual messages and thus can conflict if a node or relationship within the same label or type has a different set of properties. I would recommend to ensure all the nodes have the same set of properties. For example, in your example, you could treat Edit: The main reason why Kafka Connect Neo4j Source Connector doesn’t support the Avro enum type because Kafka Connect schema doesn't support Enum type. |
Description
When attempting to use the Neo4j Kafka Connect Source connector configured with an AVRO value converter and corresponding schema in Schema Registry, the connector is unable to correctly produce records due to serialization errors.
Expected Behavior (Mandatory)
When the connector is configured as Source in QUERY mode with a valid AVRO schema, the messages should be correctly serialized and produced to the configured topic.
Actual Behavior (Mandatory)
The connector throws an error due to it being unable to parse messages properly with the configured AVRO schema.
While testing different configurations, a few issues surfaced:
When the AVRO schema has fields configured as
enum, and an incoming message has a valid value for the field, the connector fails withorg.apache.avro.AvroTypeException: Not an enum: {EXPECTED_VALUE} for schema.When disabling the
neo4j.enforce.schemaoption, and having a valid incoming message, the connector fails withjava.lang.ClassCastException: class java.lang.String cannot be cast to class org.apache.avro.generic.IndexedRecord.When one of the returned fields by the Cypher query is
null, the connector fails withorg.apache.kafka.connect.errors.SchemaBuilderException: fieldSchema for field {FIELD_NAME} cannot be null.How to Reproduce the Problem
Dataset and configurations
Connector config
{ "name": "Neo4jSourceConnectorAVRO", "config": { "topic": "neo4j.test.avro", "connector.class": "streams.kafka.connect.source.Neo4jSourceConnector", "key.converter": "org.apache.kafka.connect.storage.StringConverter", "key.converter.schemas.enable": "false", "value.converter": "io.confluent.connect.avro.AvroConverter", "value.converter.schema.registry.url": "http://schema-registry:8081", "value.converter.use.latest.version": "true", "value.converter.enhanced.avro.schema.support": "true", "value.converter.latest.compatibility.strict": "false", "value.converter.auto.register.schemas": "false", "value.converter.connect.meta.data": "true", "neo4j.server.uri": "bolt://neo4j:7687", "neo4j.authentication.basic.username": "neo4j", "neo4j.authentication.basic.password": "password", "neo4j.streaming.poll.interval.msecs": 5000, "neo4j.streaming.from": "LAST_COMMITTED", "neo4j.enforce.schema": "true", "neo4j.source.query": "MATCH (n:TestNode) WHERE n.lastModifiedAt > datetime({epochMillis: toInteger($lastCheck)}) RETURN n.uuid AS uuid, n.name AS name, n.company AS company, n.numericId AS numericId, n.entityType AS entityType, n.status AS status, n.isDeleted AS isDeleted;" } }AVRO schema
{ "type": "record", "name": "myNode", "namespace": "com.mycompany.neo4j.avro", "fields": [ { "name": "uuid", "type": { "type": "string", "logicalType": "uuid" }, "doc": "Mandatory node UUID." }, { "name": "name", "type": "string", "doc": "Mandatory node Name." }, { "name": "company", "type": ["null", "string"], "doc": "Optional node Company." }, { "name": "numericId", "type": "int" }, { "name": "entityType", "type": { "name": "entityTypeTypes", "type": "enum", "symbols": [ "Company", "Subcompany" ] }, "doc": "Mandatory node entityType." }, { "name": "status", "type": [ "null", { "type": "enum", "name": "statusTypes", "symbols": [ "approved", "reviewed", "deleted" ] } ], "doc": "Optional node status." }, { "name": "isDeleted", "type": "boolean", "doc": "Mandatory node isDeleted." } ] }Sample test Cypher to produce records for the connector
Steps (Mandatory)
uuidlogicalTypeTest cases and stacktraces
This are the results when:
Companyis not part of the expected ENUMS ofCompany,Subcompanyfor theentityTypefield:neo4j.enforce.schematofalse, it produces a different stack trace and fails:companyin the node tonull, andneo4j.enforce.schemaistrue, the connector fails and produces:neo4j.enforce.schemaset tofalse, a different error happens:Specifications and Versions (Mandatory)
The initial issues with the AVRO schema were found with the following specs:
confluentinc/cp-kafka-connect-base:7.3.1To replicate the issue and discard older version issues, I used the config provided above through the docker-compose file provided:
I have also tested the schema separately with other connectors and with a custom Producer using the schema without any issues.
The text was updated successfully, but these errors were encountered: