用 Neo4j Kafka 连接器与原生 CDC API…

在 Kafka 与 Neo4j 之间做实时数据流（sink 写入、source 变更捕获、或不用 Kafka 直接查询变更事件）时，选择并配置正确的连接器策略与 CDC 游标循环。

适用场景

解决思路

1. 按用例选择连接器策略：自定义转换 Kafka 载荷→图用 Sink Cypher；镜像另一个 Neo4j CDC 源用 Sink CDC（schema 或 source-id 子策略）；无代码把 Kafka JSON 字段映射为节点/关系用 Sink Pattern；消费预格式化 CUD JSON 消息用 Sink CUD；实时把 Neo4j 全部变更流到 Kafka 用 Source CDC（需 Neo4j 5.13+ EE/Aura BC/VDC）；按计划流式输出特定查询结果用 Source Query；在进程内消费 CDC 事件不用 Kafka 用原生 CDC API（db.cdc.query）。
2. 确认前置条件：Neo4j Connector for Kafka ≥ 5.5.2（5.5.2 按分区而非按 topic 处理消息，5.5.1 会在生成查询中强制静态标签）、Kafka Connect ≥ 3.x 或 Confluent Platform ≥ 7.x、CDC source/sink 需 Neo4j 5.13+ Enterprise/AuraDB Business Critical/AuraDB VDC、query source 任意版本、Java 11+。
3. 所有连接器共用连接配置：设置 neo4j.uri（如 neo4j+s://your-instance.databases.neo4j.io:7687）、neo4j.authentication.type（BASIC|BEARER|KERBEROS|CUSTOM|NONE）、neo4j.authentication.basic.username、neo4j.authentication.basic.password、neo4j.database；密码绝不要硬编码，用 Kafka Connect secrets provider（${file:...} 或 ${env:...}）。
4. Sink Cypher 策略：连接器会自动前置 UNWIND $events AS __value，查询中用 __value 引用消息；配置 neo4j.cypher.topic.<topic> 为每个 topic 指定 Cypher，可用 neo4j.cypher.bind-value-as、neo4j.cypher.bind-key-as、neo4j.cypher.bind-header-as 自定义绑定名；用 MERGE 做幂等 upsert，例如 MERGE (p:Person {id: __value.id}) ON CREATE SET p.createdAt = datetime(), p += __value.properties ON MATCH SET p.updatedAt = datetime(), p += __value.properties。
5. Sink Pattern 策略：无需 Cypher，用模式语法映射消息字段到图，如 neo4j.pattern.topic.users = (:User{!userId, name, email})、neo4j.pattern.topic.friendships = (:User{!userId: from.userId})-[:KNOWS{since}]->(:User{!userId: to.userId})；规则：!prop 为键属性（用于 MERGE）、prop: field.path 从嵌套字段映射、* 映射全部字段、-prop 排除属性（不能与包含混用）。
6. Sink CDC 策略：镜像另一个 Neo4j 时配置 neo4j.cdc.schema.topics；或用 source-id 跟踪（把 elementId 存为属性）配置 neo4j.cdc.source-id.topics、neo4j.cdc.source-id.label-name、neo4j.cdc.source-id.property-name。
7. 启用 exactly-once 语义（EOS）：需连接器 ≥ 5.3.0（Cypher/Pattern/CDC 策略）、≥ 5.3.1（CUD 策略）、Kafka broker 支持 EOS，并先创建 NODE KEY 约束：CREATE CONSTRAINT kafka_offset_key IF NOT EXISTS FOR (n:__KafkaOffset) REQUIRE (n.strategy, n.topic, n.partition) IS NODE KEY；然后在连接器配置加 neo4j.eos-offset-label: __KafkaOffset；未启用 EOS 时为 at-least-once，需写幂等 Cypher（用 MERGE 而非 CREATE）。
8. 配置错误处理与 DLQ：生产环境设置 errors.tolerance=all、errors.log.enable=true、errors.log.include.messages=true、errors.deadletterqueue.topic.name=neo4j-dlq、errors.deadletterqueue.context.headers.enable=true、errors.deadletterqueue.topic.replication.factor=3；默认 errors.tolerance=none 会在首个错误处停止。
9. Source CDC 策略（推荐，Neo4j 5.13+）：配置 connector.class 为 org.neo4j.connectors.kafka.source.Neo4jConnector、neo4j.source-strategy=CDC、neo4j.start-from（NOW|EARLIEST|具体游标字符串）、neo4j.cdc.poll-interval、neo4j.cdc.poll-duration，并按 topic 配置模式与操作，如 neo4j.cdc.topic.person-creates.patterns.0.pattern=(:Person)、neo4j.cdc.topic.person-creates.patterns.0.operation=CREATE；每个 topic 可配多个模式，索引 0、1、2…；注意数据库从备份恢复后 CDC 游标失效，需重新配置 neo4j.start-from。
10. Source Query 策略（旧版/任意版本）：配置 neo4j.source-strategy=QUERY、neo4j.query（连接器自动注入 $lastCheck）、neo4j.query.streaming-property（必须由查询返回且应建索引）、neo4j.query.topic、neo4j.query.polling-interval、neo4j.query.polling-duration。
11. 使用原生 CDC API（无需 Kafka，需 Neo4j 5.13+ Enterprise/AuraDB BC/VDC）：自管实例在 neo4j.conf 设 db.cdc.enabled=true，Aura 合格层级默认开启；用 CALL db.cdc.current() YIELD id RETURN id AS cursor 获取当前游标，或用 CALL db.cdc.earliest() 获取最早可用游标（游标是排他的，current() 不包含其指向的事务）。
12. 用 db.cdc.query 查询变更：CALL db.cdc.query($cursor, []) YIELD id, txId, seq, metadata, event RETURN ... ORDER BY txId, seq；可按选择器过滤，如 [{select: 'n', labels: ['Person'], operation: 'c'}] 或 [{select: 'r', type: 'KNOWS', changesTo: ['since', 'strength']}]。
13. 掌握选择器字段：select（'e' 全部/'n' 节点/'r' 关系）、operation（'c'/'u'/'d'）、labels（节点须含全部标签）、type（关系类型）、elementId、key（需键约束）、changesTo（AND，全部须变更）、authenticatedUser、executingUser、txMetadata。
14. 理解事件结构：顶层含 id（该事件的游标，用作下一个 $cursor）、txId、seq、metadata（executingUser、authenticatedUser、captureMode 为 DIFF 或 FULL、txStartTime、txCommitTime、txMetadata）、event（elementId、eventType 为 n 或 r、operation 为 c/u/d、labels、type、keys、state.before/after.properties，CREATE 时 before 为 null，DELETE 时 after 为 null）。
15. 实现游标循环（Python）：用 driver.execute_query 执行 CALL db.cdc.query($cursor, $selectors) YIELD id, txId, seq, event RETURN ... ORDER BY txId, seq；把游标推进到最后一个事件的 id，无事件时保持当前游标；启动时用 CALL db.cdc.current() 引导游标；循环中处理事件后 sleep 再轮询；仅在成功处理后推进游标。
16. 配置 Confluent Cloud 托管连接器：无需上传 JAR，无 connector.class 字段（在 UI/API 选择）；凭据用 Confluent Cloud secret manager 或直接 JSON；支持私有端点（AWS PrivateLink、Azure Private Link、GCP PSC）；托管升级，必要时显式固定连接器版本；必填字段包括 kafka.auth.mode=KAFKA_API_KEY、kafka.api.key、kafka.api.secret、input.data.format=JSON、neo4j.uri、neo4j.authentication.type、neo4j.authentication.basic.username、neo4j.authentication.basic.password；每个 topic 只能选一种策略，不能在同一 topic 混用 Cypher 和 Pattern。
17. 配置 Schema Registry（Avro/JSON Schema）：source 连接器总是生成带 schema 支持的消息，必须配置转换器，如 key.converter/value.converter 为 io.confluent.connect.avro.AvroConverter 并设 schema.registry.url；JSON Schema 用 io.confluent.connect.json.JsonSchemaConverter；sink 转换器必须与 source 匹配（Avro sink 不能消费 JSON schema source 消息）。
18. 按常见错误排查：CDC is not enabled→在 neo4j.conf 启用或升级到 EE/BC/VDC；数据库恢复后 Invalid cursor→把 neo4j.start-from 重置为 NOW 或 EARLIEST；Cannot merge node using null→校验消息 schema 并在 Cypher 加空值检查；重启后消息重放→加 neo4j.eos-offset-label 与 NODE KEY 约束；坏消息导致连接器停止→设 errors.tolerance=all 加 DLQ；sink 报 SchemaException→两端匹配 key/value 转换器；db.cdc.query 返回空事件→用 db.cdc.earliest() 重放或等待新事务。
19. 按检查清单核对：确认 CDC 可用性（Neo4j 5.13+ EE/Aura BC/VDC）；sink 导入前创建唯一性/NODE KEY 约束（MERGE 依赖它们）；使用 neo4j.eos-offset-label 时创建 EOS 约束；凭据走 secrets provider 不硬编码；Cypher sink 查询用 MERGE 而非 CREATE 保证幂等；生产 sink 配置 errors.tolerance=all 加 DLQ；source 的 neo4j.query.streaming-property 建索引；source 与 sink 两侧 schema registry 转换器匹配；数据库恢复后重配 CDC 游标；CDC 游标循环仅在成功处理后推进游标。

适用边界 · 注意事项

• 不处理 Cypher 查询编写——应使用 neo4j-cypher-skill。
• 不处理批量 CSV/文件导入——应使用 neo4j-import-skill。
• 不处理 GDS 算法——应使用 neo4j-gds-skill；实时应用写入模式也应使用 neo4j-cypher-skill。