Quickstart: Source with Query

When to use the Query strategy

Use the Query strategy when you want to publish a specific slice of the graph and you can express "what changed since last time" as a Cypher query. The connector runs your query on an interval, passing in a cursor as $lastCheck.

Pick it when:

you are on Community Edition, or cannot enable Change Data Capture;
you only want a narrow projection, not every change to the data;
you want to shape the message in the query — aggregate, join, rename, filter;
your data already has a monotonically increasing tracking property.

This strategy cannot detect deletes. A query only returns rows that exist, so a deleted node simply stops appearing — there is no event for it. If you need deletes, either use CDC or adopt soft deletes: keep the node, set a deleted flag, and bump the tracking property so the query picks up the change.

What you will build

One source connector polling a query and publishing each returned row to a topic:

(:Reading) nodes ──every 1s──> MATCH ... WHERE r.timestamp > $lastCheck ──> readings topic

Prerequisites

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

Community Edition is fine — this strategy needs neither CDC nor Enterprise features.

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.

Step 1: Index the tracking property

CREATE INDEX reading_timestamp IF NOT EXISTS FOR (r:Reading) ON (r.timestamp);

The connector filters on this property every poll. Without an index that is a full label scan on each cycle, which gets slower as the data grows.

Step 2: Create the topic

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

Step 3: Create the source connector

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

{
  "name": "Neo4jSourceQueryQuickstart",
  "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": "QUERY",
    "neo4j.start-from": "EARLIEST",
    "neo4j.query.topic": "readings",
    "neo4j.query": "MATCH (r:Reading) WHERE r.timestamp > $lastCheck RETURN r.sensor AS sensor, r.celsius AS celsius, r.timestamp AS timestamp ORDER BY r.timestamp",
    "neo4j.query.streaming-property": "timestamp",
    "neo4j.query.poll-interval": "1s",
    "neo4j.query.poll-duration": "5s"
  }
}

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

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

How the query works

MATCH (r:Reading)
WHERE r.timestamp > $lastCheck
RETURN r.sensor AS sensor, r.celsius AS celsius, r.timestamp AS timestamp
ORDER BY r.timestamp

$lastCheck: Supplied by the connector on every poll. It starts from neo4j.start-from and then advances to the tracking property of the last row of each batch.
neo4j.query.streaming-property: Names the returned field used as the cursor — timestamp here. It must be part of the RETURN, or the connector cannot advance.
ORDER BY: Not optional in practice. The connector takes the cursor from the last row it received, so if rows come back unordered it can jump past values it never published.

The tracking property must be an integer, not a datetime. The cursor is handled internally as a long — neo4j.start-from: NOW seeds it with System.currentTimeMillis() — so the property should hold epoch milliseconds, which is what Cypher’s timestamp() returns. Returning a DATETIME instead fails with Returned record does not contain a valid field timestamp … expected a long value.

Note also that the comparison is >, strictly greater than. Two rows sharing a timestamp value can mean the second is never published, so a tracking property with sufficient resolution matters. Milliseconds is usually enough; seconds often is not.

Where polling starts

neo4j.start-from is EARLIEST in this configuration, which seeds the cursor with -1 so existing rows are published on the first poll. That makes the walkthrough below easy to follow.

For a real deployment NOW is usually what you want — it seeds the cursor with the current timestamp, so only new rows are published. USER_PROVIDED takes an explicit value in neo4j.start-from.value. All three apply only on first run; once Kafka Connect has stored an offset, the setting is ignored unless you set neo4j.ignore-stored-offset.

Step 4: Insert some data

CREATE (:Reading {sensor: 'kitchen', celsius: 21.5, timestamp: timestamp()});
CREATE (:Reading {sensor: 'garage', celsius: 8.0, timestamp: timestamp()});

timestamp() returns the current time in epoch milliseconds as an integer, which is exactly what the cursor expects.

Step 5: Read the messages

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

Each message is a {"schema": …, "payload": …} envelope, and each payload is one row of your query — the aliases in the RETURN become the field names:

{"sensor": "kitchen", "celsius": 21.5, "timestamp": 1730000000000}

Now add another row and watch it arrive within a poll interval:

CREATE (:Reading {sensor: 'attic', celsius: 15.25, timestamp: timestamp()});

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

Then confirm the cursor is doing its job — update an existing row without touching its timestamp:

MATCH (r:Reading {sensor: 'kitchen'}) SET r.celsius = 30.0;

Nothing new is published. The query only sees rows whose timestamp exceeds the cursor, so a change that does not bump the tracking property is invisible. Bump it and the row is published again:

MATCH (r:Reading {sensor: 'kitchen'}) SET r.celsius = 30.0, r.timestamp = timestamp();

That is the discipline this strategy requires: every write that should be published must advance the tracking property. It is the main reason to prefer CDC when it is available.

Message shape and types

Payload Mode controls how values are represented:

neo4j.payload-mode Effect

`neo4j.payload-mode`	Effect
`EXTENDED`	Default. Rich type information preserved, at the cost of a more nested message.
`COMPACT`	Simpler messages, some type fidelity lost.
`RAW_JSON_STRING`	The whole payload as a single JSON string, for consumers that parse it themselves.

EXTENDED

Default. Rich type information preserved, at the cost of a more nested message.

COMPACT

Simpler messages, some type fidelity lost.

RAW_JSON_STRING

The whole payload as a single JSON string, for consumers that parse it themselves.

Returning whole nodes rather than properties also works — RETURN r — but the message then carries Neo4j-specific structure. Projecting explicit fields, as above, keeps consumers independent of your graph model.

Troubleshooting

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

Returned record does not contain a valid field <prop> … expected a long value: The tracking property is not an integer, or is missing from the RETURN. Use timestamp(), not datetime().
Topic stays empty: With neo4j.start-from: NOW only rows created after startup qualify. Check whether your rows predate the connector, and that their timestamp really exceeds the cursor.
Rows published repeatedly: The query returns rows whose tracking property does not advance past the cursor. Confirm the WHERE uses $lastCheck and that neo4j.query.streaming-property names a returned field.
Some rows never appear: Either two rows share a tracking value — the comparison is strictly > — or the query has no ORDER BY on the tracking property, so the cursor skipped ahead.
Deletes not appearing: Expected. This strategy cannot see deletes; use soft deletes or CDC.
Polling gets slower over time: The tracking property is not indexed. See Step 1.

Next steps

Source Query strategy — full reference.
Quickstart: Source with CDC — the alternative, when you need deletes.
Payload Mode — message representation.
Source settings — batching, polling, offsets.
Monitoring — JMX metrics exposed by the source connector.