Neo4j Kafka源连接器遇null字段无法构建Schema求助

问题重现 #

使用Neo4j Kafka源连接器同步带null值的age字段时，AvroConverter和JsonSchemaConverter均报错，仅StringConverter可正常工作。连接器核心配置如下：

{
    "connector.class": "streams.kafka.connect.source.Neo4jSourceConnector",
    "neo4j.server.uri": "bolt://neo4j:7687",
    "neo4j.source.query": "MATCH (c:Customer) WHERE c.timestamp > $lastCheck RETURN c.name as name, c.age as age, c.timestamp as timestamp",
    "neo4j.enforce.schema": "true",
    "topic": "neo4j-test-AVRO",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "neo4j.streaming.poll.interval.msecs": "5000",
    "neo4j.streaming.from": "LAST_COMMITTED"
}

插入含nullage的节点时触发报错：

CREATE (c1:Customer {name: 'Test',age:null, timestamp: timestamp()})

原因分析 #

Avro和JSON Schema转换器依赖严格的类型校验与Schema兼容性：

1. 当neo4j.enforce.schema=true时，连接器会根据首次同步的数据生成固定Schema。如果首次同步的age字段非null，Schema会被定义为非空的数值类型；后续出现null值时，与已注册的Schema不兼容，触发报错。
2. StringConverter仅做字符串序列化，不校验类型与Schema，因此可忽略null值的类型冲突。

解决方案 #

方案1：修改Cypher查询，明确支持null字段 #

调整查询语句，确保连接器生成的Schema包含nullable标记：

MATCH (c:Customer) WHERE c.timestamp > $lastCheck 
RETURN c.name as name,
       c.age AS age,  -- 显式保留null，让连接器识别字段可空
       c.timestamp as timestamp

或使用COALESCE给null值设置默认值（适合允许非null默认值的场景）：

MATCH (c:Customer) WHERE c.timestamp > $lastCheck 
RETURN c.name as name,
       COALESCE(c.age, -1) as age,  -- 将null替换为-1，避免Schema冲突
       c.timestamp as timestamp

方案2：调整连接器与转换器配置 #

1. 关闭强制Schema校验，让连接器动态生成包含nullable字段的Schema：
   将neo4j.enforce.schema修改为false。
2. 配置AvroConverter支持null类型：
   添加以下配置项：

   "value.converter.connect.meta.data": "true",
   "value.converter.avro.nullable": "true"
   

方案3：预注册兼容的Avro Schema #

在Schema Registry中预先注册包含可空age字段的Schema，确保后续数据匹配：

{
  "type": "record",
  "name": "Customer",
  "namespace": "com.example",
  "fields": [
    {"name": "name", "type": "string"},
    {"name": "age", "type": ["null", "int"]},  -- 明确标记age为可空类型
    {"name": "timestamp", "type": "long"}
  ]
}

然后配置连接器使用该预注册Schema（需确保转换器配置指向正确的Schema ID或名称）。

内容的提问来源于stack exchange，提问作者sharmajee499