Configure and operate the Neo4j Connector for Kafka (sink + source) and the native Neo4j CDC API. Covers Cypher/Pattern/CUD sink strategies, CDC-based and query-based source, exactly-once semantics, DLQ error handling, Confluent Cloud managed connector, schema registry (Avro/JSON), and native db.cdc

Neo4j Kafka Skill When to Use Writing Kafka events into Neo4j (sink connector — Cypher, Pattern, CDC, CUD strategies) Streaming Neo4j changes to Kafka topics (source connector — CDC or query based) Querying Neo4j change events natively without Kafka ( db.cdc.query ) Configuring Confluent Cloud managed Neo4j sink connector Setting up schema registry (Avro/JSON Schema) for typed Kafka messages Enabling exactly once semantics or dead letter queue on sink When NOT to Use Cypher query authoring → neo4j cypher skill Bulk CSV/JSON file import → neo4j import skill GDS algorithms → neo4j gds skill Live app write patterns → neo4j cypher skill Decision Table — Which connector strategy? Use case Strategy Custom transformation of Kafka payload → graph Sink: Cypher Mirror another Neo4j CDC source Sink: CDC (schema or source id sub strategy) Map Kafka JSON fields to graph nodes/rels with no code Sink: Pattern Consume pre formatted CUD JSON messages Sink: CUD Stream all Neo4j changes to Kafka (real time) Source: CDC (Neo4j 5.13+ EE/Aura BC/VDC) Stream specific query results on a schedule Source: Query Consume CDC events in process, no Kafka Native CDC API ( db.cdc.query ) Prerequisites Neo4j Connector for Kafka ≥ 5.5.2 (download from [neo4j.com/labs/kafka](https://neo4j.com/labs/kafka/) or Confluent Hub) — 5.5.2 processes incoming messages per partition instead of per topic; 5.5.1 forces static labels in generated queries Kafka Connect ≥ 3.x or Confluent Platform ≥ 7.x For CDC source/sink: Neo4j 5.13+ Enterprise Edition, AuraDB Business Critical, or AuraDB VDC For query source: any Neo4j edition Java 11+ Core Connection Config (all connectors) Authentication types: BASIC BEARER KERBEROS CUSTOM NONE Never hardcode passwords — use Kafka Connect secrets provider ( ${file:...} or ${env:...} ). Sink Connector Strategy 1 — Cypher Connector auto prepends UNWIND $events AS value — write query using value : MERGE pattern — idempotent upsert: Strategy 2 — Pattern No Cypher needed — map message fields to graph via pattern syntax: Pattern rules: !prop = key property (used for MERGE) prop: field.path = map from nested message field = map all message fields prop = exclude property (cannot mix with inclusions) Strategy 3 — CDC (mirror another Neo4j) Or with source id tracking (stores elementId as property): Exactly Once Semantics (EOS) Requires: connector ≥ 5.3.0 (Cypher/Pattern/CDC strategies), connector ≥ 5.3.1 (CUD strategy), Kafka broker EOS support, and a NODE KEY constraint. Step 1 — Create constraint: Step 2 — Add to connector config: Without EOS: connector provides at least once — write idempotent Cypher (MERGE, not CREATE). Error Handling / DLQ errors.tolerance=none (default) — stops on first error. Use all + DLQ for production. Source Connector CDC Based Source (recommended, Neo4j 5.13+) neo4j.start from options: NOW EARLIEST a specific cursor string Multiple patterns per topic — indexed 0, 1, 2...: Cursor warning: after DB restore from backup, CDC cursors are invalidated. Reconfigure neo4j.start from . Query Based Source (legacy / any edition) $lastCheck is auto injected by connector. neo4j.query.streaming property must be returned by the query and should be indexed. Native CDC API (no Kafka required) Requires: Neo4j 5.13+ Enterprise, AuraDB BC, or AuraDB VDC. Enable CDC first (self managed — set in neo4j.conf): On Aura: enabled by default on eligible tiers. Cursor Bootstrap Cursors are exclusive: db.cdc.current() does NOT include the transaction it points to. Query Changes Filtered — nodes with label Person, CREATE only: Filtered — specific relationship type with property change tracking: Selector Reference Field Values Applies to select 'e' (all), 'n' (nodes), 'r' (rels) both operation 'c' (create), 'u' (update), 'd' (delete) both labels ['Label1','Label2'] (node must have ALL) nodes type 'REL TYPE' relationships elementId specific element ID string both key {propName: value} (requires key constraint) both changesTo ['prop1','prop2'] (AND — all must change) both authenticatedUser username string both executingUser username string both txMetadata {key: value} both Event Structure Cursor Loop Pattern (Python) Confluent Cloud Managed Connector Confluent Cloud hosts the Neo4j Sink connector as a fully managed service (no JAR upload needed). Config differences vs self managed: No connector.class field — selected in UI/API Credentials via Confluent Cloud secret manager or direct JSON Private endpoints supported (AWS PrivateLink, Azure Private Link, GCP PSC) Managed upgrades — pin connector version explicitly if needed Required Confluent Cloud fields: One strategy per topic — cannot mix Cypher and Pattern on same topic. Schema Registry (Avro / JSON Schema) Source connector always generates messages with schema support — must configure converters: For JSON Schema: Sink converter must match source — Avro sink cannot consume JSON schema source messages. Common Errors Error Cause Fix CDC is not enabled db.cdc.enabled not set / wrong tier Enable in neo4j.conf or upgrade to EE/BC/VDC Invalid cursor after DB restore Backup invalidates cursors Reset neo4j.start from to NOW or EARLIEST Cannot merge node using null Key property missing in message Validate message schema; add null check in Cypher Messages replayed after restart No EOS configured Add neo4j.eos offset label + NODE KEY constraint Connector stops on bad message errors.tolerance=none (default) Set errors.tolerance=all + DLQ topic SchemaException on sink Converter mismatch source/sink Match key/value converters on both ends Empty events from db.cdc.query Cursor points to current Use db.cdc.earliest() to replay; wait for new txns References [Full connector config reference](references/sink config.md) — all neo4j. properties, defaults, types [CDC API patterns](references/cdc api.md) — cursor loop, selector examples, event structure detail [Neo4j Connector for Kafka docs](https://neo4j.com/docs/kafka/current/) [CDC docs](https://neo4j.com/docs/cdc/current/) Checklist [ ] CDC availability confirmed (Neo4j 5.13+ EE / Aura BC / VDC) if using CDC source or sink [ ] Uniqueness/NODE KEY constraints created before sink import (MERGE uses them) [ ] EOS constraint created if using neo4j.eos offset label [ ] Credentials via secrets provider — not hardcoded in config [ ] Cypher sink queries use MERGE (not CREATE) for idempotency [ ] errors.tolerance=all + DLQ configured for production sink [ ] Source: neo4j.query.streaming property indexed [ ] Schema registry converters match on both source and sink sides [ ] After DB restore: CDC cursor reconfigured ( neo4j.start from ) [ ] CDC cursor loop: advance cursor only after successful processing

neo4j-kafka-skill