Quickstart: Sink with Change Data Capture

When to use the CDC strategy

Use the Change Data Capture strategy when the messages you are consuming were produced by Neo4j — either by a source connector using the CDC strategy or by the deprecated Neo4j Streams plugin. The connector understands the change-event format natively and reapplies each change to the target database.

Pick it when you are:

replicating a Neo4j database into another Neo4j or Aura database;
migrating between instances with Kafka in the middle;
fanning one database out to several downstream Neo4j instances.

Unlike the other sink strategies, this one cannot be driven by hand-written messages. A change event must arrive as a schema-carrying structure and with the neo4j.source.cdc.id header that the source connector attaches; without the header the connector treats the message as the legacy Neo4j Streams format instead. So this guide runs a real source connector, which is also the only realistic use case.

What you will build

Two connectors and two databases, with Kafka in between:

source database ──> Source connector (CDC) ──> cdc-events topic ──> Sink connector ──> target database

Two separate databases are essential. If the sink wrote back into the source database, its writes would be captured by CDC, republished, and applied again — an endless loop.

This guide uses JsonConverter with schemas.enable=true, which embeds the schema in every message. That satisfies the schema requirement without needing Schema Registry, keeping the setup small. Avro and Protobuf work the same way — see Using Avro or Protobuf.

Prerequisites

Docker Compose v2.20.3 or later.
The Neo4j Connector for Kafka archive, unpacked as described in Installation.
Neo4j Enterprise Edition or Aura Enterprise. CDC and multiple databases are not available in Community Edition.

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 two databases

In Neo4j Browser at http://localhost:7474 (neo4j / password), against the system database:

CREATE DATABASE source IF NOT EXISTS;
CREATE DATABASE target IF NOT EXISTS;

Step 2: Enable CDC on the source database only

ALTER DATABASE source SET OPTION txLogEnrichment 'FULL';

Leave target alone.

For more on enabling CDC, see Change Data Capture > Enable CDC > Neo4j DBMS for on-prem installations and Change Data Capture > Enable CDC > Aura for Aura.

On Aura, CDC is reset to OFF when an instance is paused and resumed. Re-enable it and follow CDC cursor recovery before restarting the source connector.

Step 3: Add key constraints on both databases

The schema sub-strategy merges entities using the key properties named in each change event. Those names travel with the event, but the constraints themselves do not — so create matching ones on both sides.

Against source and again against target:

CREATE CONSTRAINT person_name IF NOT EXISTS
FOR (p:Person) REQUIRE (p.first_name, p.last_name) IS NODE KEY;

On source this constraint is what puts first_name and last_name into the event’s keys field. On target it gives the generated MERGE an index to use and enforces uniqueness.

Without a constraint on source, change events carry no key properties and the schema sub-strategy has nothing to merge on. Without one on target, writes still succeed but lose both uniqueness and index-backed lookup.

Step 4: Create the source connector

Save the following as source.cdc.neo4j.json:

{
  "name": "Neo4jSourceCdcQuickstart",
  "config": {
    "connector.class": "org.neo4j.connectors.kafka.source.Neo4jConnector",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": true,
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": true,
    "neo4j.uri": "neo4j://neo4j:7687",
    "neo4j.authentication.type": "BASIC",
    "neo4j.authentication.basic.username": "neo4j",
    "neo4j.authentication.basic.password": "password",
    "neo4j.database": "source",
    "neo4j.source-strategy": "CDC",
    "neo4j.start-from": "NOW",
    "neo4j.cdc.poll-interval": "1s",
    "neo4j.cdc.poll-duration": "5s",
    "neo4j.cdc.topic.cdc-events.patterns": "(:Person),(:Person)-[:KNOWS]->(:Person)"
  }
}

Both patterns publish to the same cdc-events topic — one for :Person nodes, one for :KNOWS relationships between them. neo4j.start-from: NOW means only changes made after the connector starts are captured.

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

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

Step 5: Create the sink connector

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

{
  "name": "Neo4jSinkCdcSchemaQuickstart",
  "config": {
    "topics": "cdc-events",
    "connector.class": "org.neo4j.connectors.kafka.sink.Neo4jConnector",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": true,
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": true,
    "neo4j.uri": "neo4j://neo4j:7687",
    "neo4j.authentication.type": "BASIC",
    "neo4j.authentication.basic.username": "neo4j",
    "neo4j.authentication.basic.password": "password",
    "neo4j.database": "target",
    "neo4j.cdc.schema.topics": "cdc-events"
  }
}

neo4j.cdc.schema.topics selects the schema sub-strategy, and neo4j.database points the writes at target. The converters match the source connector’s exactly — this is not optional, as the sink has to reconstruct the change events the source serialized.

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

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

Both connectors and both tasks should report RUNNING before you continue.

Step 6: Make some changes and watch them replicate

Against the source database:

CREATE (:Person {first_name: 'John', last_name: 'Doe', email: '[email protected]'});
CREATE (:Person {first_name: 'Mary', last_name: 'Doe', email: '[email protected]'});
MATCH (j:Person {first_name: 'John'}), (m:Person {first_name: 'Mary'})
CREATE (j)-[:KNOWS {since: date('2012-01-01')}]->(m);

Now switch to the target database — :use target in Browser — and read it back:

MATCH (p:Person) RETURN p.first_name AS first, p.last_name AS last, p.email AS email ORDER BY first

first    last    email
"John"   "Doe"   "[email protected]"
"Mary"   "Doe"   "[email protected]"

MATCH (a:Person)-[k:KNOWS]->(b:Person) RETURN a.first_name AS from, b.first_name AS to, k.since AS since

Updates and deletes replicate too. Back on source:

MATCH (p:Person {first_name: 'John'}) SET p.email = '[email protected]';
MATCH (p:Person {first_name: 'Mary'}) DETACH DELETE p;

On target, John’s email is updated and Mary is gone, along with the :KNOWS relationship.

The two sub-strategies

Step 5 used the schema sub-strategy. There is a second one, and the difference is what each merges on.

neo4j.cdc.schema.topics neo4j.cdc.source-id.topics

	`neo4j.cdc.schema.topics`	`neo4j.cdc.source-id.topics`
Merges on	the key properties named in the event, from the source database’s constraints	the source database’s internal `elementId`
Needs constraints	yes, on both databases	no
Target graph	natural keys, no extra properties	an extra label and property holding the source id
Use when	the target should look like the source, and you have keys	there are no usable keys, or you want a faithful id-for-id copy

Merges on

the key properties named in the event, from the source database’s constraints

the source database’s internal elementId

Needs constraints

yes, on both databases

Target graph

natural keys, no extra properties

an extra label and property holding the source id

Use when

the target should look like the source, and you have keys

there are no usable keys, or you want a faithful id-for-id copy

To try the Source ID sub-strategy, delete the sink connector and register one that swaps the strategy key:

curl -X DELETE http://localhost:8083/connectors/Neo4jSinkCdcSchemaQuickstart

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

Every replicated entity then also carries a :SourceEvent label and a sourceId property holding the source database’s element id. Both extra settings are optional; the defaults are shown above.

Using Avro or Protobuf

The requirement is a converter that carries schemas — plain JSON with schemas.enable=false cannot work here, because the connector needs a structured value rather than a plain map.

JsonConverter with schemas.enable=true is used above because it needs no extra infrastructure. To use the Schema Registry included in the Compose stack instead, set the same converter on both connectors:

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

Avro and Protobuf also preserve temporal and numeric types more faithfully than embedded-schema JSON. See Schema Registry and Type support.

Troubleshooting

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

Confirm events are actually reaching Kafka before blaming the sink:

docker compose exec broker kafka-console-consumer \
  --bootstrap-server broker:29092 --topic cdc-events --from-beginning --max-messages 1

Nothing in the topic: CDC is not enabled on source, or the changes predate the connector. neo4j.start-from is NOW, so only changes after startup are captured. Verify with SHOW DATABASE source YIELD name, options.
unexpected message value type: The sink is reading with a schemaless converter. Both connectors need schemas.enable=true, or Avro/Protobuf.
Events arrive but nothing is written, no error: The value converter carries no neo4j.source.cdc.id header, so the message is being parsed as the legacy Neo4j Streams format. Check that the messages really came from a Neo4j source connector.
Nodes duplicated on the target: No key constraint on target, so MERGE had nothing unique to match. See Step 3.
Nodes merged into one on the target: The source constraint covers fewer properties than you expected, so several source nodes share one key. Check the constraint on source.
Writes loop endlessly: CDC is enabled on target as well. Turn it off with ALTER DATABASE target SET OPTION txLogEnrichment 'OFF'.

Next steps

Change Data Capture strategy — both sub-strategies in full, including how each change type maps onto Cypher.
Source CDC strategy — pattern syntax, operations and metadata filters.
Source best practices — cursor recovery and sizing.
Error handling — dead letter queues and error tolerance.
Monitoring — JMX metrics exposed by the sink connector.