Quickstart: Sink with Pattern

When to use the Pattern strategy

Use the Pattern strategy when your messages map onto graph elements field-for-field, and you would rather declare that mapping than write a query. You give a Cypher-like pattern per topic; the connector generates the MERGE for you.

Pick it when:

each message describes one node, or one relationship with its two endpoints;
the properties you want are already fields on the message, possibly renamed or nested;
you want key-based upserts without hand-writing MERGE clauses.

One topic carries exactly one pattern. You cannot extract two node types, or a node and a relationship, from the same topic — use a separate topic per pattern. This guide therefore uses two topics.

What you will build

Two connectors' worth of mapping in a single connector instance: a node pattern on one topic and a relationship pattern on another.

users topic     ──>  (:User {userId, name, surname})

purchases topic ──>  (:User {userId})-[:BOUGHT {price, currency}]->(:Product {productId, name})

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.

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

Step 1: Create key constraints

Patterns upsert on their key properties — the fields marked with !. Create a matching node key constraint for each, in Neo4j Browser:

CREATE CONSTRAINT user_id IF NOT EXISTS FOR (u:User) REQUIRE u.userId IS NODE KEY;
CREATE CONSTRAINT product_id IF NOT EXISTS FOR (p:Product) REQUIRE p.productId IS NODE KEY;

This step is optional but recommended. Without it the connector still works, but it logs a warning per pattern — Label 'User' does not match the key(s) defined by the pattern … — and every upsert falls back to an unindexed scan.

Node key constraints require Neo4j Enterprise Edition or Aura. On Community Edition, use REQUIRE u.userId IS UNIQUE instead.

Step 2: Create the topics

One topic per pattern:

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

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

Step 3: Create the sink connector

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

{
  "name": "Neo4jSinkPatternQuickstart",
  "config": {
    "topics": "users,purchases",
    "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.pattern.merge-node-properties": true,
    "neo4j.pattern.topic.users": "(:User{!userId, name, surname})",
    "neo4j.pattern.topic.purchases": "(:User{!userId})-[:BOUGHT{price, currency}]->(:Product{!productId, name: productName})"
  }
}

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

Confirm it is running:

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

Both the connector and its task should report RUNNING.

Reading the two patterns

The node pattern on the users topic:

(:User{!userId, name, surname})

:User — the label to merge on.
!userId — a key property. MERGE matches on it, so re-sending the same userId updates the existing node rather than creating a second one. At least one key property is required.
name, surname — the only other properties copied onto the node. Any other field in the message is ignored.

The relationship pattern on the purchases topic:

(:User{!userId})-[:BOUGHT{price, currency}]->(:Product{!productId, name: productName})

both endpoints are merged on their key properties, then :BOUGHT is merged between them;
{price, currency} are copied onto the relationship;
name: productName renames a message field on the way in — the productName field becomes the name property on :Product.

Two shorthands worth knowing: {!userId, *} (or just {!userId}) copies all message fields, and {!userId, -address} copies all except address. You cannot mix inclusion and exclusion in one pattern. Full syntax is in Pattern strategy.

Note "neo4j.pattern.merge-node-properties": true in the configuration. It is required here, and this is the setting most likely to surprise you.

Both topics write to the same :User nodes. On a relationship pattern, an endpoint like (:User{!userId}) contributes no value properties — wildcards are not allowed on endpoints — so with the default false the connector issues the equivalent of SET u = {} before re-applying the key. That erases the name and surname written by the users topic, and the node is left holding only userId.

Set it to true and the connector uses SET u += … instead, leaving properties written by other topics intact. As a rule: whenever more than one topic writes to the same node, you want this enabled. neo4j.pattern.merge-relationship-properties does the same for relationship properties.

Step 4: Produce messages

Create the connector before producing, so nothing is missed. Users first:

docker compose exec -T broker kafka-console-producer \
  --bootstrap-server broker:29092 \
  --topic users <<'EOF'
{"userId": 1, "name": "John", "surname": "Doe"}
{"userId": 2, "name": "Mary", "surname": "Smith"}
EOF

Then purchases:

docker compose exec -T broker kafka-console-producer \
  --bootstrap-server broker:29092 \
  --topic purchases <<'EOF'
{"userId": 1, "productId": 100, "productName": "Laptop", "price": 999.99, "currency": "USD"}
{"userId": 2, "productId": 100, "productName": "Laptop", "price": 949.00, "currency": "USD"}
{"userId": 2, "productId": 200, "productName": "Keyboard", "price": 49.99, "currency": "USD"}
EOF

Note that productId: 100 appears twice. Because productId is a key property, both messages merge onto the same :Product node.

Step 5: Verify the result

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

MATCH (u:User)-[b:BOUGHT]->(p:Product)
RETURN u.name AS user, p.name AS product, b.price AS price ORDER BY user, product

user     product      price
"John"   "Laptop"     999.99
"Mary"   "Keyboard"   49.99
"Mary"   "Laptop"     949.0

Confirm the merge behaviour — two users, two products, three relationships:

MATCH (n) RETURN labels(n) AS item, count(*) AS count
UNION ALL
MATCH ()-[r]->() RETURN [type(r)] AS item, count(*) AS count

Now re-send one of the purchase messages:

docker compose exec -T broker kafka-console-producer \
  --bootstrap-server broker:29092 \
  --topic purchases <<'EOF'
{"userId": 1, "productId": 100, "productName": "Laptop", "price": 899.00, "currency": "USD"}
EOF

Re-run the first query. John’s :BOUGHT price is updated to 899.0 and no new node or relationship is created — that is the key properties doing their work.

Deleting with tombstones

The Pattern strategy also handles tombstone records: a message whose key carries the pattern’s key properties and whose value is null deletes the matching element. This needs a producer that can emit a null value and a structured key, so it is not shown here — see Tombstone records.

Using Avro, Protobuf or JSON Schema

Patterns are unchanged across formats; only the converters differ. Replace the value.converter lines with, for example:

  "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. Schema-carrying formats also preserve Kafka Connect logical types — with plain JSON, a date arrives as a string, whereas Avro and Protobuf map it onto a Neo4j temporal type. See Type support.

Troubleshooting

curl -s http://localhost:8083/connectors/Neo4jSinkPatternQuickstart/status
docker compose logs connect

Message value must be convertible to a Map: The Pattern strategy requires an object-shaped message value. A bare scalar or array will not work — unlike the Cypher strategy, which can bind any value.
Task fails with a pattern parse error: The pattern is validated at startup. Check that every node pattern has at least one ! key property, and that inclusion and exclusion properties are not mixed.
Label 'X' does not match the key(s) defined by the pattern: A warning, not an error. Create the node key constraint from Step 1.
Nodes created but properties missing: The pattern lists properties explicitly. A field absent from the pattern is not copied — use {!key, *} to copy everything.
A node’s properties vanish after a second topic writes to it: neo4j.pattern.merge-node-properties defaults to false, which makes the connector replace a node’s whole property map rather than add to it. A relationship endpoint contributes no value properties, so it replaces them with nothing. Set the option to true. Diagnose it with MATCH (n:User) RETURN n.userId, keys(n) — you will see only the key property left.
Nothing at all happens: Confirm each topic in topics has a matching neo4j.pattern.topic.<topic> entry. A topic with no pattern is consumed and silently ignored.

Next steps

Pattern strategy — full pattern grammar, tombstones, batching and exactly-once semantics.
Sink settings — neo4j.pattern.bind-*, batching, retries.
Error handling — dead letter queues and error tolerance.
Monitoring — JMX metrics exposed by the sink connector.