Quickstart: Sink with Cypher

When to use the Cypher strategy

Use the Cypher strategy when you consume messages from a Kafka topic and want full control over the write query that applies them to Neo4j. Each message is passed to a Cypher statement you provide, as a parameter.

Pick it when you need to:

compute or derive property values, rather than copy fields verbatim;
write conditional logic, run multiple MERGE clauses, or touch several labels per message;
map one message onto a subgraph whose shape does not follow directly from the message fields.

What you will build

A single sink connector that consumes JSON messages from a topic named people, and for each message merges a Person node, a City node and a LIVES_IN relationship between them.

people topic ──> Sink connector ──> (:Person)-[:LIVES_IN]->(:City)

This guide does not use a source connector — messages are produced by hand, the way an external system would produce them.

This guide uses plain JSON messages, so no Schema Registry is required. See Using Avro, Protobuf or JSON Schema for Avro, Protobuf and JSON Schema.

Prerequisites

Docker Compose v2.20.3 or later.
The Neo4j Connector for Kafka archive, unpacked as described in Installation.

Start the environment

Copy the following Docker Compose file into a new directory.

docker-compose.yml

---
services:
  neo4j:
    image: neo4j:2026-enterprise
    hostname: neo4j
    container_name: neo4j
    # this is to ensure you have the latest 2026.x version of the database
    pull_policy: always
    ports:
      - "7474:7474"
      - "7687:7687"
    environment:
      NEO4J_AUTH: neo4j/password
      NEO4J_ACCEPT_LICENSE_AGREEMENT: "yes"
      NEO4J_server_memory_heap_max__size: "4G"
    healthcheck:
      test: [ "CMD", "cypher-shell", "-u", "neo4j", "-p", "password", "RETURN 1" ]
      start_period: 2m
      start_interval: 10s
      interval: 30s
      timeout: 10s
      retries: 5

  broker:
    image: confluentinc/cp-server:7.8.0
    hostname: broker
    container_name: broker
    ports:
      - "9092:9092"
      - "9101:9101"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: 'broker,controller'
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@broker:29093'
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
      KAFKA_LISTENERS: 'PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092'
      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
      CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
      KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CONFLUENT_BALANCER_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: localhost
      KAFKA_CONFLUENT_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: broker:29092
      CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
      CONFLUENT_METRICS_ENABLE: 'true'
      CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous'
    healthcheck:
      test: [ "CMD", "nc", "-z", "localhost", "9092" ]
      start_period: 5m
      start_interval: 10s
      interval: 1m
      timeout: 10s
      retries: 5

  schema-registry:
    image: confluentinc/cp-schema-registry:7.8.0
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
      - broker
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:29092'
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
    healthcheck:
      test: [ "CMD", "nc", "-z", "localhost", "8081" ]
      start_period: 5m
      start_interval: 10s
      interval: 1m
      timeout: 10s
      retries: 5

  connect:
    image: confluentinc/cp-server-connect:7.8.0
    hostname: connect
    container_name: connect
    depends_on:
      - broker
      - schema-registry
    ports:
      - "8083:8083"
    volumes:
      - ./plugins:/tmp/connect-plugins
    environment:
      CONNECT_BOOTSTRAP_SERVERS: 'broker:29092'
      CONNECT_REST_ADVERTISED_HOST_NAME: connect
      CONNECT_GROUP_ID: compose-connect-group
      CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000
      CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter
      CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      # CLASSPATH required due to CC-2422
      CLASSPATH: /usr/share/java/monitoring-interceptors/monitoring-interceptors-7.8.0.jar
      CONNECT_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor"
      CONNECT_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor"
      CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components,/tmp/connect-plugins"
      CONNECT_LOG4J_LOGGERS: org.apache.zookeeper=ERROR,org.I0Itec.zkclient=ERROR,org.reflections=ERROR
    healthcheck:
      test: [ "CMD", "nc", "-z", "localhost", "8083" ]
      start_period: 5m
      start_interval: 10s
      interval: 1m
      timeout: 10s
      retries: 5

  control-center:
    image: confluentinc/cp-enterprise-control-center:7.8.0
    hostname: control-center
    container_name: control-center
    depends_on:
      - broker
      - schema-registry
      - connect
    ports:
      - "9021:9021"
    environment:
      CONTROL_CENTER_BOOTSTRAP_SERVERS: 'broker:29092'
      CONTROL_CENTER_CONNECT_CONNECT-DEFAULT_CLUSTER: 'connect:8083'
      CONTROL_CENTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      CONTROL_CENTER_REPLICATION_FACTOR: 1
      CONTROL_CENTER_INTERNAL_TOPICS_PARTITIONS: 1
      CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS: 1
      CONFLUENT_METRICS_TOPIC_REPLICATION: 1
      PORT: 9021
    healthcheck:
      test: [ "CMD", "curl", "-f", "http://localhost:9021" ]
      start_period: 5m
      start_interval: 10s
      interval: 1m
      timeout: 10s
      retries: 5

Copy the Neo4j Connector for Kafka JAR into a directory named plugins next to your docker-compose.yml, so that the directory looks like this:

quickstart/
├─ plugins/
│  ├─ neo4j-kafka-connect-5.5.2.jar
├─ docker-compose.yml

Start the stack:

docker compose up -d

Wait until every service reports as healthy — this takes 90-120 seconds on a first run:

docker compose ps

Confirm that the connector plugin was loaded:

curl -s http://localhost:8083/connector-plugins | grep -o 'org.neo4j.connectors.kafka.[a-z]*.Neo4jConnector'

Both …source.Neo4jConnector and …sink.Neo4jConnector should be listed. Empty output means the JAR is missing from plugins or failed to load — check docker compose logs connect.

You can now log in to Neo4j Browser at http://localhost:7474 with the username neo4j and the password password.

The Cypher strategy does not require Change Data Capture, so there is nothing to enable on the database.

Step 1: Create the topic

docker compose exec broker kafka-topics \
  --bootstrap-server broker:29092 \
  --create --topic people --partitions 1 --replication-factor 1

Step 2: Create the sink connector

Save the following as sink.cypher.neo4j.json:

{
  "name": "Neo4jSinkCypherQuickstart",
  "config": {
    "topics": "people",
    "connector.class": "org.neo4j.connectors.kafka.sink.Neo4jConnector",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": false,
    "neo4j.uri": "neo4j://neo4j:7687",
    "neo4j.authentication.type": "BASIC",
    "neo4j.authentication.basic.username": "neo4j",
    "neo4j.authentication.basic.password": "password",
    "neo4j.cypher.topic.people": "WITH __value AS person MERGE (p:Person {name: person.name, surname: person.surname}) SET p.fullName = person.name + ' ' + person.surname MERGE (c:City {name: person.city}) MERGE (p)-[:LIVES_IN]->(c)"
  }
}

The neo4j.cypher.topic.people setting is the whole strategy: it names the topic (neo4j.cypher.topic.<topic>) and gives the Cypher statement to run for its messages. Written out, that statement is:

WITH __value AS person
MERGE (p:Person {name: person.name, surname: person.surname})
SET p.fullName = person.name + ' ' + person.surname
MERGE (c:City {name: person.city})
MERGE (p)-[:LIVES_IN]->(c)

__value is the message value, bound automatically for every message. The header, key and timestamp are available as __header, __key and __timestamp. See Cypher strategy settings to rename these.

Note the SET p.fullName = … clause — deriving a value like this is the kind of thing that makes Cypher the right strategy rather than Pattern.

curl -X POST http://localhost:8083/connectors \
  -H 'Content-Type:application/json' \
  -H 'Accept:application/json' \
  -d @sink.cypher.neo4j.json

Confirm it is running:

curl -s http://localhost:8083/connectors/Neo4jSinkCypherQuickstart/status

Both the connector and its task should report RUNNING.

Step 3: Produce messages

Create the connector before producing, so that no message is missed.

docker compose exec -T broker kafka-console-producer \
  --bootstrap-server broker:29092 \
  --topic people <<'EOF'
{"name": "Jane", "surname": "Doe", "city": "London"}
{"name": "John", "surname": "Doe", "city": "London"}
{"name": "Mary", "surname": "Smith", "city": "Berlin"}
EOF

Step 4: Verify the result

In Neo4j Browser at http://localhost:7474/browser/, run:

MATCH (p:Person)-[:LIVES_IN]->(c:City) RETURN p.fullName AS person, c.name AS city ORDER BY person

You should get three rows, and two City nodes — Jane and John both merged onto the same London node:

person       city
"Jane Doe"   "London"
"John Doe"   "London"
"Mary Smith" "Berlin"

Produce another message to see it applied:

docker compose exec -T broker kafka-console-producer \
  --bootstrap-server broker:29092 \
  --topic people <<'EOF'
{"name": "Ann", "surname": "Bolin", "city": "Berlin"}
EOF

Re-run the query — Ann Bolin is added and attached to the existing Berlin node.

Using Avro, Protobuf or JSON Schema

The strategy is identical for schema-carrying formats; only the converters change. Replace the value.converter lines in the connector configuration with one of the following and point them at the Schema Registry included in the Compose stack:

  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter.schema.registry.url": "http://schema-registry:8081"

You then need a schema-aware producer, such as kafka-avro-console-producer, in place of kafka-console-producer. See Cypher strategy for configuration examples in each format, and Type support for how Kafka Connect types map onto Neo4j types.

Troubleshooting

Check the connector status first — a failed task reports its exception here:

curl -s http://localhost:8083/connectors/Neo4jSinkCypherQuickstart/status

Then check the worker log:

docker compose logs connect

No nodes created: Confirm the topic name in topics matches the suffix of neo4j.cypher.topic.<topic>. A mismatch means the connector consumes the topic but has no statement for it.
Task failed with a Cypher error: The statement is executed as written. Test it in Neo4j Browser with a sample parameter before configuring it.
Messages produced before the connector existed: Sink connectors start from the earliest available offset by default, but if you have already consumed them, produce them again.

To route failing messages to a dead letter queue instead of stopping the task, see Error handling.

Next steps

Cypher strategy — full reference, including __key and __header binding.
Sink settings — batching, retries and connection options.
Error handling — dead letter queues and error tolerance.
Monitoring — JMX metrics exposed by the sink connector.