Neo4j Kafka Skill

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?

Prerequisites

• Neo4j Connector for Kafka ≥ 5.5.2 (download from 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)

{
  "neo4j.uri": "neo4j+s://your-instance.databases.neo4j.io:7687",
  "neo4j.authentication.type": "BASIC",
  "neo4j.authentication.basic.username": "neo4j",
  "neo4j.authentication.basic.password": "${file:/opt/secrets.properties:neo4j.password}",
  "neo4j.database": "neo4j"
}

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:

{
  "connector.class": "org.neo4j.connectors.kafka.sink.Neo4jConnector",
  "topics": "person-creates,person-updates",
  "neo4j.uri": "neo4j+s://...",
  "neo4j.authentication.type": "BASIC",
  "neo4j.authentication.basic.username": "neo4j",
  "neo4j.authentication.basic.password": "secret",
  "neo4j.cypher.topic.person-creates":
    "MERGE (p:Person {id: __value.id}) SET p += __value.properties",
  "neo4j.cypher.topic.person-updates":
    "MATCH (p:Person {id: __value.id}) SET p += __value.properties",
  "neo4j.cypher.bind-value-as": "__value",
  "neo4j.cypher.bind-key-as": "__key",
  "neo4j.cypher.bind-header-as": "__header"
}

MERGE pattern — idempotent 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

Strategy 2 — Pattern

No Cypher needed — map message fields to graph via pattern syntax:

{
  "neo4j.pattern.topic.users": "(:User{!userId, name, email})",
  "neo4j.pattern.topic.friendships":
    "(:User{!userId: from.userId})-[:KNOWS{since}]->(:User{!userId: to.userId})"
}

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)

{
  "neo4j.cdc.schema.topics": "neo4j-cdc-events"
}

Or with source-id tracking (stores elementId as property):

{
  "neo4j.cdc.source-id.topics": "neo4j-cdc-events",
  "neo4j.cdc.source-id.label-name": "SourceEvent",
  "neo4j.cdc.source-id.property-name": "sourceId"
}

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:

CREATE CONSTRAINT kafka_offset_key IF NOT EXISTS
FOR (n:__KafkaOffset)
REQUIRE (n.strategy, n.topic, n.partition) IS NODE KEY;

Step 2 — Add to connector config:

{
  "neo4j.eos-offset-label": "__KafkaOffset"
}

Without EOS: connector provides at-least-once — write idempotent Cypher (MERGE, not CREATE).

Error Handling / 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 (default) — stops on first error. Use all + DLQ for production.

Source Connector

CDC-Based Source (recommended, Neo4j 5.13+)

{
  "connector.class": "org.neo4j.connectors.kafka.source.Neo4jConnector",
  "neo4j.uri": "neo4j+s://...",
  "neo4j.authentication.type": "BASIC",
  "neo4j.authentication.basic.username": "neo4j",
  "neo4j.authentication.basic.password": "secret",
  "neo4j.source-strategy": "CDC",
  "neo4j.start-from": "NOW",
  "neo4j.cdc.poll-interval": "1s",
  "neo4j.cdc.poll-duration": "5s",
  "neo4j.cdc.topic.person-creates.patterns.0.pattern": "(:Person)",
  "neo4j.cdc.topic.person-creates.patterns.0.operation": "CREATE",
  "neo4j.cdc.topic.person-updates.patterns.0.pattern": "(:Person)",
  "neo4j.cdc.topic.person-updates.patterns.0.operation": "UPDATE",
  "neo4j.cdc.topic.person-deletes.patterns.0.pattern": "(:Person)",
  "neo4j.cdc.topic.person-deletes.patterns.0.operation": "DELETE"
}

neo4j.start-from options: NOW | EARLIEST | a specific cursor string

Multiple patterns per topic — indexed 0, 1, 2...:

{
  "neo4j.cdc.topic.all-changes.patterns.0.pattern": "(:Person)",
  "neo4j.cdc.topic.all-changes.patterns.1.pattern": "(:Organization)"
}

Cursor warning: after DB restore from backup, CDC cursors are invalidated. Reconfigure neo4j.start-from.

Query-Based Source (legacy / any edition)

{
  "neo4j.source-strategy": "QUERY",
  "neo4j.query": "MATCH (p:Person) WHERE p.updatedAt > $lastCheck RETURN p.id AS id, p.name AS name, p.updatedAt AS updatedAt",
  "neo4j.query.streaming-property": "updatedAt",
  "neo4j.query.topic": "person-changes",
  "neo4j.query.polling-interval": "5s",
  "neo4j.query.polling-duration": "10s"
}

$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):

db.cdc.enabled=true

On Aura: enabled by default on eligible tiers.

Cursor Bootstrap

// Get cursor for "right now" — start tracking from this point forward
CALL db.cdc.current() YIELD id RETURN id AS cursor;

// Get earliest available cursor (replay from history start)
CALL db.cdc.earliest() YIELD id RETURN id AS cursor;

Cursors are exclusive: db.cdc.current() does NOT include the transaction it points to.

Query Changes

// All changes since cursor
CALL db.cdc.query($cursor, []) YIELD id, txId, seq, metadata, event
RETURN id, txId, seq, metadata, event
ORDER BY txId, seq;

Filtered — nodes with label Person, CREATE only:

CALL db.cdc.query($cursor, [
  {select: 'n', labels: ['Person'], operation: 'c'}
]) YIELD id, txId, seq, event
RETURN id, event.state.after.properties AS newProps
ORDER BY txId, seq;

Filtered — specific relationship type with property change tracking: