Quickstart: Sink with CUD

When to use the CUD strategy

Use the CUD strategy when each message states its own operation. Instead of you declaring a mapping once in the connector configuration, every message carries an op field telling the connector whether to create, update, merge or delete — and the connector generates the matching Cypher.

Pick it when:

  • the producer already knows the intent of each change, and deletes are part of the stream;

  • one topic must carry operations on several different labels and relationship types;

  • you want a single stream to drive a full entity lifecycle.

Unlike Pattern, CUD has no one-mapping-per-topic restriction — a single topic can carry nodes and relationships of any label. This guide uses one topic for everything.

What you will build

One connector consuming a crm topic, driving a full lifecycle:

crm topic ──> create  (:Customer), (:Product)
              create  (:Customer)-[:PURCHASED]->(:Product)
              update  (:Customer)
              merge   (:Customer)
              delete  (:Customer)

No source connector is involved — 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.

Step 1: Create the topic

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

Step 2: Create the sink connector

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

{
  "name": "Neo4jSinkCudQuickstart",
  "config": {
    "topics": "crm",
    "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.cud.topics": "crm"
  }
}

Two settings name the same topic, and both are required:

  • topics — the Kafka Connect subscription, telling the worker which topics to consume.

  • neo4j.cud.topics — telling the connector to interpret those topics as CUD messages.

A topic listed in topics but missing from neo4j.cud.topics is consumed and silently ignored.

Register the connector:

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

Both the connector and its task should report RUNNING.

The CUD message format

Every message is a JSON object. type and op are always required; the rest depends on them.

Table 1. Node messages
Field Required Meaning

type

yes

node

op

yes

create, update, merge or delete

labels

no

labels to apply, as an array

properties

except for delete

the properties to set

ids

except for create

the lookup key, as an object

detach

no

on delete, also remove attached relationships. Defaults to false
Table 2. Relationship messages
Field Required Meaning

type

yes

relationship

op

yes

create, update, merge or delete

rel_type

yes

the relationship type

from / to

yes

endpoint objects, each with labels, ids and an optional op

properties

no

the properties to set on the relationship

ids

no

lookup key for the relationship itself

op and type are case-insensitive. Inside from and to, op accepts only match (the default) or merge — it decides whether a missing endpoint is created.

kafka-console-producer reads one message per line, so each CUD object must be on a single line when produced this way. The examples in CUD File Format Strategy are pretty-printed for readability; the commands below are the same objects flattened.

Step 3: Create nodes

docker compose exec -T broker kafka-console-producer \
  --bootstrap-server broker:29092 \
  --topic crm <<'EOF'
{"type":"node","op":"create","labels":["Customer"],"properties":{"id":1,"name":"John Doe","tier":"standard"}}
{"type":"node","op":"create","labels":["Customer"],"properties":{"id":2,"name":"Mary Smith","tier":"standard"}}
{"type":"node","op":"create","labels":["Product"],"properties":{"id":100,"name":"Laptop"}}
EOF

Each of these becomes CREATE (n:Customer) SET n = $properties.

Verify in Neo4j Browser at http://localhost:7474/browser/:

MATCH (n) RETURN labels(n) AS labels, n.id AS id, n.name AS name ORDER BY id

Step 4: Create a relationship

The endpoints are looked up by their ids, so they must already exist:

docker compose exec -T broker kafka-console-producer \
  --bootstrap-server broker:29092 \
  --topic crm <<'EOF'
{"type":"relationship","op":"create","rel_type":"PURCHASED","from":{"labels":["Customer"],"ids":{"id":1}},"to":{"labels":["Product"],"ids":{"id":100}},"properties":{"price":999.99,"currency":"USD"}}
EOF

This becomes:

MATCH (start:Customer {id: $from.ids.id}) WITH start
MATCH (end:Product {id: $to.ids.id}) WITH start, end
CREATE (start)-[r:PURCHASED]->(end)
SET r = $properties

Because both endpoints default to match, a missing endpoint means the MATCH finds nothing and no relationship is created — without an error. To create endpoints on demand, set op on them:

docker compose exec -T broker kafka-console-producer \
  --bootstrap-server broker:29092 \
  --topic crm <<'EOF'
{"type":"relationship","op":"create","rel_type":"PURCHASED","from":{"labels":["Customer"],"ids":{"id":2},"op":"merge"},"to":{"labels":["Product"],"ids":{"id":200},"op":"merge"},"properties":{"price":49.99,"currency":"USD"}}
EOF

Product 200 did not exist and is created by the merge.

MATCH (c:Customer)-[r:PURCHASED]->(p:Product)
RETURN c.name AS customer, p.id AS product, r.price AS price ORDER BY customer

Step 5: Update and merge

update sets properties on an existing node, leaving others intact (SET n += $properties):

docker compose exec -T broker kafka-console-producer \
  --bootstrap-server broker:29092 \
  --topic crm <<'EOF'
{"type":"node","op":"update","labels":["Customer"],"ids":{"id":1},"properties":{"tier":"gold"}}
EOF

Customer 1’s tier becomes gold and name is untouched.

merge creates the node if the ids do not match anything, and updates it if they do:

docker compose exec -T broker kafka-console-producer \
  --bootstrap-server broker:29092 \
  --topic crm <<'EOF'
{"type":"node","op":"merge","labels":["Customer"],"ids":{"id":3},"properties":{"name":"Ann Bolin","tier":"standard"}}
{"type":"node","op":"merge","labels":["Customer"],"ids":{"id":3},"properties":{"tier":"gold"}}
EOF

Customer 3 is created by the first message and updated by the second — one node, not two.

MATCH (c:Customer) RETURN c.id AS id, c.name AS name, c.tier AS tier ORDER BY id

Step 6: Delete

A delete needs only ids. Deleting a node that still has relationships fails unless you ask for a detach delete:

docker compose exec -T broker kafka-console-producer \
  --bootstrap-server broker:29092 \
  --topic crm <<'EOF'
{"type":"node","op":"delete","labels":["Customer"],"ids":{"id":3}}
{"type":"node","op":"delete","labels":["Customer"],"ids":{"id":2},"detach":true}
EOF

Customer 3 has no relationships, so a plain delete works. Customer 2 bought product 200, so it needs "detach": true.

MATCH (c:Customer) RETURN c.id AS id, c.name AS name ORDER BY id

Only customer 1 is left, confirming both deletes applied.

delete is for individual entities identified by ids. It is not a way to run arbitrary Cypher deletes from JSON.

Using Avro, Protobuf or JSON Schema

The CUD format is unchanged across serialization formats; only the converters differ:

  "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. Note that CUD messages are deeply nested and loosely typed by nature, which makes them awkward to express as a strict Avro or Protobuf schema — plain JSON is often the more natural fit for this strategy. See Type support.

Troubleshooting

curl -s http://localhost:8083/connectors/Neo4jSinkCudQuickstart/status
docker compose logs connect
Message value must be convertible to a Map

The message was not a JSON object — check for a stray blank line or a pretty-printed object split across several lines, since kafka-console-producer treats each line as its own message.

Failed to transform message

The object parsed but is not a valid operation. Check that type is node or relationship, that op is one of create, update, merge, delete, and that ids is present for everything except create.

Relationship silently not created

The endpoints default to op: match. Either create the endpoint nodes first, or set "op": "merge" inside from and to.

Delete appears to do nothing

ids did not match any entity. Confirm the property name and the value type — {"id": 1} and {"id": "1"} are different lookups.

Task fails on a node delete

The node still has relationships. Add "detach": true.

Nothing happens at all

The topic is in topics but missing from neo4j.cud.topics.

Next steps

  • CUD File Format Strategy — every field, every operation, and the exact Cypher each one generates, including _id and _elementId lookups.

  • Sink settings — batching, retries and connection options.

  • Error handling — dead letter queues, useful here since a malformed operation is a per-message failure.

  • Monitoring — JMX metrics exposed by the sink connector.