Neo4j CDC事件属性值格式异常排查及简化方案咨询

问题 #

已在Neo4j实例（版本5.22）中启用CDC，并配置Kafka Connect（版本5.1.1）监听数据库数据变更。根据官方文档，事件数据应为键值对格式，但实际收到的每个属性都带有B、I64、S等额外字段的复杂结构。

收到的Kafka主题示例数据：

{  
  "id": "CJUg4WrNW0Y7ttlh8lbkxfwAAAAAAAAEMAAAAAAAAAACAAABkh21o5c=",  
  "txId": 1072,  
  "seq": 2,  
  "event": {    
    "elementId": "4:9520e16a-cd5b-463b-b6d9-61f256e4c5fc:2073",    
    "eventType": "NODE",    
    "operation": "CREATE",    
    "labels": [      
      "Environment"    
    ],    
    "keys": {},    
    "state": {      
      "before": null,      
      "after": {        
        "labels": [          
          "Environment"        
        ],        
        "properties": {          
          "name": {            
            "B": null,            
            "I64": null,            
            "F64": null,            
            "S": "Dev",            
            "BA": null,            
            "TLD": null,            
            "TLDT": null,            
            "TLT": null,            
            "TZDT": null,            
            "TOT": null,            
            "TD": null,            
            "SP": null,            
            "LB": null,            
            "LI64": null,            
            "LF64": null,            
            "LS": null,            
            "LTLD": null,            
            "LTLDT": null,            
            "LTLT": null,            
            "LZDT": null,            
            "LTOT": null,            
            "LTD": null,            
            "LSP": null          
          },          
          "id": {            
            "B": null,            
            "I64": null,            
            "F64": null,            
            "S": "78b90e78-9b79-4330-9d02-7895f349964b",            
            "BA": null,            
            "TLD": null,            
            "TLDT": null,            
            "TLT": null,            
            "TZDT": null,            
            "TOT": null,            
            "TD": null,            
            "SP": null,            
            "LB": null,            
            "LI64": null,            
            "LF64": null,            
            "LS": null,            
            "LTLD": null,            
            "LTLDT": null,            
            "LTLT": null,            
            "LZDT": null,            
            "LTOT": null,            
            "LTD": null,            
            "LSP": null          
        }        
      }      
    }    
  }  
}

当前Kafka连接器配置：

{  
  "name": "neo4j-source-connector",  
  "config": {    
    "connector.class": "org.neo4j.connectors.kafka.source.Neo4jConnector",    
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",    
    "key.converter.schemas.enable": false,    
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",    
    "value.converter.schemas.enable": false,    
    "neo4j.uri": "bolt://localhost:7687",    
    "neo4j.streaming.from": "ALL",    
    "neo4j.authentication.basic.username": "neo4j",    
    "neo4j.authentication.basic.password": "password",    
    "neo4j.source-strategy": "CDC",    
    "neo4j.start-from": "NOW",    
    "neo4j.cdc.poll-interval": "500ms",    
    "neo4j.cdc.poll-duration": "5s",    
    "neo4j.cdc.topic.neo4j-node-topic.patterns.0.pattern": "(:Environment)"  
  }  
}

期望得到的是简单键值格式（如"id": "78b90e78-9b79-4330-9d02-7895f349964b"），而非带类型标识的复杂结构，询问是否缺少配置及如何简化。

解决方案 #

这种带类型字段的结构是Neo4j CDC事件的原始格式，它通过不同字段标识属性的类型（比如S代表字符串、I64代表64位整数）。要简化为普通键值对，需要在连接器配置中添加以下关键参数：

• 替换值转换器并启用解包
  将原来的org.apache.kafka.connect.json.JsonConverter替换为Neo4j专属转换器，并开启解包配置：

  "value.converter": "org.neo4j.connectors.kafka.converter.Neo4jJsonConverter",
  "value.converter.unwrap": true
  

  这个转换器会自动解析带类型标识的属性结构，输出普通键值对格式。

• 确认版本兼容性
  Neo4jJsonConverter是Neo4j Kafka Connector自带组件，确保连接器版本与Neo4j 5.22匹配，避免版本差异导致功能失效。

• 修改后的完整配置

  {  
    "name": "neo4j-source-connector",  
    "config": {    
      "connector.class": "org.neo4j.connectors.kafka.source.Neo4jConnector",    
      "key.converter": "org.apache.kafka.connect.storage.StringConverter",    
      "key.converter.schemas.enable": false,    
      "value.converter": "org.neo4j.connectors.kafka.converter.Neo4jJsonConverter",    
      "value.converter.schemas.enable": false,    
      "value.converter.unwrap": true,    
      "neo4j.uri": "bolt://localhost:7687",    
      "neo4j.streaming.from": "ALL",    
      "neo4j.authentication.basic.username": "neo4j",    
      "neo4j.authentication.basic.password": "password",    
      "neo4j.source-strategy": "CDC",    
      "neo4j.start-from": "NOW",    
      "neo4j.cdc.poll-interval": "500ms",    
      "neo4j.cdc.poll-duration": "5s",    
      "neo4j.cdc.topic.neo4j-node-topic.patterns.0.pattern": "(:Environment)"  
    }  
  }
  

修改配置后重启Kafka Connect，收到的事件数据中properties部分会变成预期格式：

"properties": {
  "name": "Dev",
  "id": "78b90e78-9b79-4330-9d02-7895f349964b"
}

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