Quickstart: Source with Change Data Capture

When to use the CDC strategy

Use the Change Data Capture strategy when you want Kafka to receive every change to the data you care about — creates, updates and deletes — as it happens, without modifying your graph model to support it.

Pick it when:

you need deletes. This is the big one: a polling query cannot see a node that is gone.
you want changes in near real time rather than on a polling interval;
you do not want to add tracking properties to your data;
you need before-and-after state for updates.

Sending these events into another Neo4j database is a separate exercise, because the sink side needs its own configuration and a second database. See Quickstart: Sink with Change Data Capture for that. This page stops at Kafka, where any consumer can read the events.

What you will build

One source connector, publishing changes to three topics split by operation:

                          ┌──> creates topic
(:Person) changes ──CDC──>├──> updates topic
                          └──> deletes topic

Routing by operation is a configuration choice, not a requirement — one topic for everything is equally valid. Splitting them makes the event shapes easier to compare.

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 is 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.

The source connector always produces messages with schemas, so a schemaless converter will not work. This guide uses JsonConverter with schemas.enable=true, which embeds the schema in each message and needs no Schema Registry. See Using Avro or Protobuf for Avro and Protobuf.

Step 1: Enable CDC

Against the system database:

ALTER DATABASE neo4j SET OPTION txLogEnrichment 'FULL';

Confirm it took effect:

SHOW DATABASES YIELD name, options WHERE name = 'neo4j' RETURN name, options

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

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

Step 2: Add a key constraint

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

This is optional for publishing, but it changes what the events contain: the constraint’s properties are what populate each event’s keys field, which is how a consumer identifies the entity without relying on Neo4j’s internal ids. Any consumer doing upserts will want it.

Step 3: Create the topics

docker compose exec broker kafka-topics --bootstrap-server broker:29092 --create --topic creates --partitions 1 --replication-factor 1
docker compose exec broker kafka-topics --bootstrap-server broker:29092 --create --topic updates --partitions 1 --replication-factor 1
docker compose exec broker kafka-topics --bootstrap-server broker:29092 --create --topic deletes --partitions 1 --replication-factor 1

Step 4: Create the source connector

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

{
  "name": "Neo4jSourceCdcOperationsQuickstart",
  "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.source-strategy": "CDC",
    "neo4j.start-from": "NOW",
    "neo4j.cdc.poll-interval": "1s",
    "neo4j.cdc.poll-duration": "5s",
    "neo4j.cdc.topic.creates.patterns.0.pattern": "(:Person)",
    "neo4j.cdc.topic.creates.patterns.0.operation": "CREATE",
    "neo4j.cdc.topic.updates.patterns.0.pattern": "(:Person)",
    "neo4j.cdc.topic.updates.patterns.0.operation": "UPDATE",
    "neo4j.cdc.topic.deletes.patterns.0.pattern": "(:Person)",
    "neo4j.cdc.topic.deletes.patterns.0.operation": "DELETE"
  }
}

Three settings deserve attention:

neo4j.cdc.topic.<topic>.patterns.N.pattern: What to capture, in Cypher-like pattern syntax. (:Person) means all changes to nodes with that label. The N is a 0-based index, so one topic can watch several patterns.
neo4j.cdc.topic.<topic>.patterns.N.operation: Narrows the pattern to CREATE, UPDATE or DELETE. Omit it to receive all three.
neo4j.start-from: NOW captures only changes made after the connector starts. EARLIEST starts from the beginning of the available change log, and USER_PROVIDED takes an explicit offset in neo4j.start-from.value. It applies only on first run — once an offset is stored in Kafka Connect, it is ignored.

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/Neo4jSourceCdcOperationsQuickstart/status

Both the connector and its task must report RUNNING before the next step — with neo4j.start-from: NOW, anything written earlier is never captured.

A source connector always runs a single task, so tasks.max above 1 has no effect.

Step 5: Make some changes

Against the neo4j database:

CREATE (:Person {first_name: 'John', last_name: 'Doe', email: '[email protected]'});
CREATE (:Person {first_name: 'Mary', last_name: 'Doe', email: '[email protected]'});

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

MATCH (p:Person {first_name: 'Mary'}) DELETE p;

Step 6: Read the events

Each topic should now hold exactly one kind of event:

docker compose exec broker kafka-console-consumer --bootstrap-server broker:29092 --topic creates --from-beginning --max-messages 2 --timeout-ms 15000

docker compose exec broker kafka-console-consumer --bootstrap-server broker:29092 --topic updates --from-beginning --max-messages 1 --timeout-ms 15000

docker compose exec broker kafka-console-consumer --bootstrap-server broker:29092 --topic deletes --from-beginning --max-messages 1 --timeout-ms 15000

With schemas.enable=true each message is a {"schema": …, "payload": …} envelope. The interesting half is the payload, whose shape is the CDC change event schema:

{
  "id": "A3Qc5ZIZ_Eo5v5xsONVo8KUAAAAAAAAADAAAAAAAAAAA",
  "seq": 0,
  "txId": 12,
  "metadata": {
    "authenticatedUser": "neo4j",
    "captureMode": "FULL",
    "connectionClient": "192.168.65.1:46246",
    "connectionServer": "172.17.0.2:7687",
    "connectionType": "bolt",
    "databaseName": "neo4j",
    "executingUser": "neo4j",
    "serverId": "7528cb82",
    "txCommitTime": "2024-03-03T20:51:56.769Z",
    "txMetadata": {
      "app": "cypher-shell_v5.6.0",
      "type": "user-direct"
    },
    "txStartTime": "2024-03-03T20:51:56.714Z"
  },
  "event": {
    "elementId": "4:741ce592-19fc-4a39-bf9c-6c38d568f0a5:0",
    "eventType": "n",
    "operation": "c",
    "keys": {
      "Person": [
        {
          "first_name": "John",
          "last_name": "Doe"
        }
      ]
    },
    "labels": [
      "Person"
    ],
    "state": {
      "before": null,
      "after": {
        "labels": [
          "Person"
        ],
        "properties": {
          "email": "[email protected]",
          "first_name": "John",
          "last_name": "Doe"
        }
      }
    }
  }
}

The fields worth knowing:

event.operation: c, u or d.
event.eventType: n for a node, r for a relationship.
event.keys: the key properties, per label, taken from the constraint you created in Step 2.
event.state.before / event.state.after: the entity before and after the change. before is null for creates, after is null for deletes — which is what lets a consumer handle deletes at all.
txId and seq: transaction id and sequence within it, giving a total ordering.

Each message also carries a neo4j.source.cdc.id Kafka header. The sink connector’s CDC strategy uses that header to recognise change events, which is why hand-written messages cannot drive it.

Capturing relationships

The pattern syntax covers relationships too. Add another pattern to any topic:

  "neo4j.cdc.topic.creates.patterns.1.pattern": "(:Person)-[:KNOWS]->(:Person)",
  "neo4j.cdc.topic.creates.patterns.1.operation": "CREATE"

Relationship events use eventType: "r" and carry start and end objects with the endpoint keys, rather than a labels field.

Patterns can also filter on changed properties and transaction metadata — see Change Data Capture strategy for the full syntax.

Choosing what goes in the message key

By default the whole change event is used as the message key, which is rarely what you want. neo4j.cdc.topic.<topic>.key-strategy accepts:

Value Key becomes

Value	Key becomes
`WHOLE_VALUE`	the entire change event. The default.
`ENTITY_KEYS`	the key properties from the event — usually the right choice for partitioning and log compaction.
`ELEMENT_ID`	the source database’s internal element id.
`SKIP`	no key at all.

WHOLE_VALUE

the entire change event. The default.

ENTITY_KEYS

the key properties from the event — usually the right choice for partitioning and log compaction.

ELEMENT_ID

the source database’s internal element id.

SKIP

no key at all.

ENTITY_KEYS is what you want if consumers need all changes to one entity on the same partition. See Best practices for the trade-offs.

Using Avro or Protobuf

The source connector always emits schemas, so the only question is which schema format. To use the Schema Registry included in the Compose stack:

  "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"

Reading those topics then needs a schema-aware consumer such as kafka-avro-console-consumer. Avro and Protobuf also represent temporal and numeric types more precisely than embedded-schema JSON. See Schema Registry and Payload Mode, which controls how property values are represented.

Troubleshooting

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

Topics stay empty: Most often CDC is not enabled, or the changes predate the connector. Check SHOW DATABASES YIELD name, options for txLogEnrichment: FULL, and remember that neo4j.start-from: NOW ignores everything written before startup.
Task fails on startup with a pattern error: Patterns are validated when the connector starts. Indexes must begin at 0 and be contiguous, and you cannot mix patterns.N.pattern with the single-string patterns form for the same topic.
Events appear in the wrong topic: Each topic’s patterns are independent. A pattern with no operation receives creates, updates and deletes.
keys is empty in every event: No constraint on the captured label. Add one as in Step 2.
Connector runs but lags behind: Raise neo4j.batch-size, or shorten neo4j.cdc.poll-interval. See Best practices.
Everything replays after a restart: Kafka Connect stores the CDC cursor as a source offset. If it is lost, neo4j.start-from applies again.

Next steps

Source CDC strategy — full pattern syntax, operations, metadata filters.
Quickstart: Sink with CDC — apply these events to another Neo4j database.
Best practices — cursor recovery, sizing, key strategies.
Payload Mode — how property values are serialized.
Monitoring — JMX metrics exposed by the source connector.