Query strategy
Query strategy allows users to define their own Cypher query to extract changes. This requires proper schema modifications, such as tracking changes through a dedicated change tracking property such as timestamps on nodes or relationships or using soft-deletes to track deletion of entities.
Configuration
First, you need to select QUERY strategy for the connector instance;
"neo4j.source-strategy": "QUERY"Second, you need to define your query to track changes and where to publish them.
"neo4j.query.topic": "my-topic", (1)
"neo4j.query": "MATCH (ts:TestSource) WHERE ts.timestamp > $lastCheck RETURN ts.name AS name, ts.surname AS surname, ts.timestamp AS timestamp", (2)
"neo4j.query.streaming-property": "timestamp" (3)| 1 | Topic name which will receive the message. |
| 2 | A Cypher query that returns changed entities since the last iteration, sent in by $lastCheck parameter. |
| 3 | The property (field name) that we use as a cursor to track changes. This needs to be part of the returned results. |
|
For good source query performance, the property used for tracking (for example, a If the connector filters or orders by this property, missing indexes can lead to full scans and significantly slower polling, especially as the dataset grows. Using a properly typed tracking property together with an index (or a suitable constraint that creates one) helps Neo4j locate new or updated records much more efficiently. |
Creating Source instance
Based on the above example, you can use one of the following configurations.
Pick one of the message serialization format examples and save it as a file named source.query.neo4j.json into a local directory.
{
"name": "Neo4jSourceConnectorAVRO",
"config": {
"connector.class": "org.neo4j.connectors.kafka.source.Neo4jConnector",
"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",
"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": "NOW",
"neo4j.query": "MATCH (ts:TestSource) WHERE ts.timestamp > $lastCheck RETURN ts.name AS name, ts.surname AS surname, ts.timestamp AS timestamp",
"neo4j.query.streaming-property": "timestamp",
"neo4j.query.topic": "test-source",
"neo4j.query.poll-interval": "1s",
"neo4j.query.poll-duration": "5s"
}
}{
"name": "Neo4jSourceConnectorJSONSchema",
"config": {
"connector.class": "org.neo4j.connectors.kafka.source.Neo4jConnector",
"key.converter": "io.confluent.connect.json.JsonSchemaConverter",
"key.converter.schemas.enable": true,
"key.converter.schema.registry.url": "http://schema-registry:8081",
"value.converter": "io.confluent.connect.json.JsonSchemaConverter",
"value.converter.schemas.enable": true,
"value.converter.schema.registry.url": "http://schema-registry:8081",
"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": "NOW",
"neo4j.query": "MATCH (ts:TestSource) WHERE ts.timestamp > $lastCheck RETURN ts.name AS name, ts.surname AS surname, ts.timestamp AS timestamp",
"neo4j.query.streaming-property": "timestamp",
"neo4j.query.topic": "test-source",
"neo4j.query.poll-interval": "1s",
"neo4j.query.poll-duration": "5s"
}
}{
"name": "Neo4jSourceConnectorProtobuf",
"config": {
"connector.class": "org.neo4j.connectors.kafka.source.Neo4jConnector",
"key.converter": "io.confluent.connect.protobuf.ProtobufConverter",
"key.converter.schemas.enable": true,
"key.converter.schema.registry.url": "http://schema-registry:8081",
"key.converter.optional.for.nullables": true,
"value.converter": "io.confluent.connect.protobuf.ProtobufConverter",
"value.converter.schemas.enable": true,
"value.converter.schema.registry.url": "http://schema-registry:8081",
"value.converter.optional.for.nullables": 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": "NOW",
"neo4j.query": "MATCH (ts:TestSource) WHERE ts.timestamp > $lastCheck RETURN ts.name AS name, ts.surname AS surname, ts.timestamp AS timestamp",
"neo4j.query.streaming-property": "timestamp",
"neo4j.query.topic": "test-source",
"neo4j.query.poll-interval": "1s",
"neo4j.query.poll-duration": "5s"
}
}{
"name": "Neo4jSourceConnectorJSONString",
"config": {
"connector.class": "org.neo4j.connectors.kafka.source.Neo4jConnector",
"key.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable": false,
"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.source-strategy": "QUERY",
"neo4j.start-from": "NOW",
"neo4j.query": "MATCH (ts:TestSource) WHERE ts.timestamp > $lastCheck RETURN ts.name AS name, ts.surname AS surname, ts.timestamp AS timestamp",
"neo4j.query.streaming-property": "timestamp",
"neo4j.query.topic": "test-source",
"neo4j.query.poll-interval": "1s",
"neo4j.query.poll-duration": "5s"
}
}We will now create the source instance by invoking the following REST call:
curl -X POST http://localhost:8083/connectors \
-H "Content-Type:application/json" \
-H "Accept:application/json" \
-d @source.query.neo4j.jsonThis will create a Kafka Connect source instance that will send change event messages derived by the provided query over to the my-topic topic, using your preferred serialization format.
In Control Center, confirm that the Source connector has been created in the Connect tab, under connect-default.
Generated change event messages can be structured in two ways, controlled by the neo4j.payload-mode setting.
The simpler structure, COMPACT, only includes the essential fields:
{
"name": "<name>",
"surname": "<surname>",
"timestamp": <timestamp>
}COMPACT is easier to read and to deserialize on the consumer side, but it does not tolerate a property changing type in Neo4j over time.
If your Cypher query returns properties whose type can change, configure neo4j.payload-mode as EXTENDED (the default) so that change event messages instead wrap each field into a dedicated, type-safe structure that stays forward compatible across property type changes:
{
"name": {
"type": "S",
"B": null,
"I64": null,
"F64": null,
"S": "<name>",
"BA": null,
"TLD": null,
"TLDT": null,
"TLT": null,
"TZDT": null,
"TOT": null,
"TD": null,
"SP": null,
"LB": null,
"LI64": null,
"LF64": null,
"LS": null,
"LTLD": null,
"LTLDT": null,
"LTLT": null,
"LZDT": null,
"LTOT": null,
"LTD": null,
"LSP": null
},
"surname": {
"type": "S",
"B": null,
"I64": null,
"F64": null,
"S": "<surname>",
"BA": null,
"TLD": null,
"TLDT": null,
"TLT": null,
"TZDT": null,
"TOT": null,
"TD": null,
"SP": null,
"LB": null,
"LI64": null,
"LF64": null,
"LS": null,
"LTLD": null,
"LTLDT": null,
"LTLT": null,
"LZDT": null,
"LTOT": null,
"LTD": null,
"LSP": null
},
"timestamp": {
"type": "I64",
"B": null,
"I64": <timestamp>,
"F64": null,
"S": null,
"BA": null,
"TLD": null,
"TLDT": null,
"TLT": null,
"TZDT": null,
"TOT": null,
"TD": null,
"SP": null,
"LB": null,
"LI64": null,
"LF64": null,
"LS": null,
"LTLD": null,
"LTLDT": null,
"LTLT": null,
"LZDT": null,
"LTOT": null,
"LTD": null,
"LSP": null
}
}The trade-off is that this structure is more verbose and can make consumer-side deserialization logic more complex, so reserve it for pipelines where type stability actually matters.
In case you generate complex data structures as part of your Cypher query, you can also use the RAW_JSON_STRING payload mode which will produce a raw JSON string representation of the data without any schema compatibility guarantees.
Although the generated message will look exactly the same as the COMPACT mode, it will be encoded as a STRING type.
Exactly-once semantics
The query strategy is not exactly-once capable, and provides at-least-once delivery guarantees. Because messages are derived from a user-provided Cypher query rather than from the transaction log, the connector cannot guarantee that each record is delivered exactly once across connector restarts or failures, so duplicate records are possible.
If you require exactly-once delivery on the source side, use the CDC source strategy instead, which declares itself exactly-once capable on Kafka Connect clusters that support KIP-618.
Refer to the payload mode page for more information about the neo4j.payload-mode setting, including its limitations.