# Neo4j CDC Source Connector

## Description

The connector provides a source component, which listens for changes occurring in a Neo4j or Aura
database and publishes messages into Apache Kafka topics. This connector as the name suggest is only configured to read
changes using Change Data Capture feature.
The official documentation is [here](https://neo4j.com/docs/kafka/5.1-preview/).


### Neo4j access configuration

Change Data Capture is a new feature introduced with Neo4j 5.13.0 and Aura 5. For Aura we have to request Neo4j team to enable CDC plugin. For local neo4j setup, for the first time use we have to enable Change Data Capture on the source database by executing the following Cypher command.
```
ALTER DATABASE neo4j SET OPTION txLogEnrichment 'FULL';
```


### Local configuration

The easiest way to test the connector locally is to use dev Kafka cluster and local neo4j 5 (docker compose). Dev Kafka cluster is available under VPN connection.

After that you need to copy `.env.shadow` to `.env` and fill in the values.

Then run the following commands to start the connector.
```
awsume dev
aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin 086679231553.dkr.ecr.us-east-1.amazonaws.com
docker-compose up --build -d
```

If all is configured correctly you should see the following logs:

```
neo4j_cdc_source connector successfully updated
neo4j_cdc_source connector started
```
and at this point you can see the connector status and its config by the following links:

- http://localhost:8083/connectors/neo4j_cdc_source/tasks/0/status
- http://localhost:8083/connectors/neo4j_cdc_source/config


### QA/PROD configuration
For the instances deployed to AWS we use Lenses' Secret Provider. This extension allows the connector to get secrets from AWS Secrets Manager. The creds should be under namespace `Environment/SERVICE_NAME/NEO4J_CREDENTIALS` where value is of format:
```
{username: "test", password: "test"}
```


### Connector configuration options:
Option | Environment variable | Desc
--- | ---| ---
neo4j.ignore-stored-offset | NEO4J_IGNORE_STORED_OFFSET | One of true, false. Allows to reset the stored offset if any. Starts from the point specified by neo4j.start-from. Default: false. Useful when running locally to force it to start from NOW instead of old offeset.
neo4j.start-from | NEO4J_START_FROM | Mandatory One of EARLIEST, NOW, USER_PROVIDED.Only used on initial run of the connector instance, and ignored when there is already a stored offset in Kafka Connect. Default: NOW
neo4j.start-from.value | NEO4J_START_FROM_VALUE | For USER_PROVIDED, CDC ID of transaction to stat from.
neo4j.cdc.{topic}.key-strategy | NEO4J_KEY_STRATEGY | Default key strategy for all topics. Can be overridden per topic via the TOPIC_MAPPING pattern (see below). Values: WHOLE_VALUE, ELEMENT_ID, ENTITY_KEYS
neo4j.enforce.schema | NEO4J_ENFORCE_SCHEMA | Apply a schema to each record (Default: false)
neo4j.database | NEO4J_DATABASE_NAME | Neo4j database name to connect to. Default is set to graph.db.
neo4j.uri | NEO4J_SERVER_URI | Neo4j URI to connect to.


### Neo4j nodes/label mapping:

Provide mapping for nodes using `TOPIC_MAPPING_xx` env variable. Nodes are mapped using their labels. You can provide a list of labels, separated by `:`. Optionally you can also define properties to be used as key filters. See examples below for different combinations. All topics can be optionally prefixed by specifying the  `TOPIC_PREFIX` env.

Pattern: `TOPIC_MAPPING_ANY_MEANINGFUL_NAME=<topic.name>|<Neo4J Label Matching Pattern CASE SENSITIVE>[|<key-strategy>]`

The third part (`key-strategy`) is optional. When omitted, the value of `NEO4J_KEY_STRATEGY` is used as the default. When provided, it overrides the default for that specific topic. Valid values: `WHOLE_VALUE`, `ELEMENT_ID`, `ENTITY_KEYS`.

Examples:

- `TOPIC_MAPPING_SOUNDRECORDING="cdc.nrGraph.soundRecording|(:NrSoundRecording)"` -- Select all changes on nodes with label :NrSoundRecording and write to topic: cdc.nrGraph.soundrecording
- `TOPIC_MAPPING_OWNERSHIP_SERVICE="cdc.nrGraph.ownershipService|(:NrOwnershipService)"` -- Select all changes on nodes with label :NrOwnershipService and write to topic: cdc.nrGraph.ownershipService
- `TOPIC_MAPPING_SOUNDRECORDING="cdc.nrGraph.soundRecording|(:NrSoundRecording:NrPerformance)"` -- Select all changes on nodes with both labels :NrSoundRecording and :NrPerformance and write to topic: cdc.nrGraph.soundRecording
- `TOPIC_MAPPING_VENDOR="cdc.nrGraph.vendor|(:Vendor {id, name})"` -- Select all changes on nodes with label :Vendor but only include id, name properties in the change event and write to topic: cdc.nrGraph.vendor
- `TOPIC_MAPPING_VENDOR="cdc.nrGraph.vendor|(:Vendor {-vendorId})"` -- Select all changes on nodes with label :Vendor but exclude vendorId property in the change event and write to topic: cdc.nrGraph.vendor
- `TOPIC_MAPPING_SUBACCOUNT="cdc.nrGraph.subaccount|(:Subaccount {deleted: 'N'})"` -- Select all changes on nodes with label :Subaccount and key property deleted equals N and write to topic: cdc.nrGraph.subaccount (lower case). This example requires a NODE KEY constraint on deleted property for :Subaccount label.
- `TOPIC_MAPPING_PRODUCT="cdc.nrGraph.product|(:Product:Orchard {upc: 'ABCD12344', productCode: 1243})"` -- Select all changes on nodes with label :Product and :Orchard and key property upc equals ABCD12344 and productCode: 1243. It will write to topic: cdc.nrGraph.product. This example requires a NODE KEY constraint on upc and productCode properties either
for :Product or :Orchard label or both.
- `TOPIC_MAPPING_SOUNDRECORDING="cdc.nrGraph.soundRecording|(:NrSoundRecording)|ELEMENT_ID"` -- Same as first example but overrides the key-strategy to ELEMENT_ID for this topic.


### Neo4j relationship mapping:

Provide mapping for nodes using `TOPIC_MAPPING_xx` env variable. Relationship patterns are defined similar to Cypher relationship patterns. See examples below for different combinations. All topics can be optionally prefixed by specifying the  `TOPIC_PREFIX` env.

Pattern: `TOPIC_MAPPING_ANY_MEANINGFUL_NAME=<topic.name>|<Neo4J Relationship Matching Pattern CASE SENSITIVE>[|<key-strategy>]`

The optional third part (`key-strategy`) works the same as for node mappings (see above).

Examples:

- `TOPIC_MAPPING_MY_TOPIC="cdc.nrGraph.myTopic|()-[:INCLUDES]->()"` -- Select all changes on :INCLUDES relationships and write to topic cdc.nrGraph.myTopic.
- `TOPIC_MAPPING_MY_TOPIC=".nrGraph.myTopic|(:NrTrack)-[:INCLUDES]->(:NrProduct)"` -- Select all changes on :INCLUDES relationships with start nodes of label :NrTrack and end nodes of label :NrProduct
- `TOPIC_MAPPING_MY_TOPIC=".nrGraph.myTopic|(:NrTrack:NrOwnership)-[:INCLUDES]->(:NrProduct:NrOwnership)"` -- Select all changes on :INCLUDES relationships with start nodes of label :NrTrack and :NrOwnership and end nodes of label :NrProduct and :NrOwnership.
- `TOPIC_MAPPING_MY_TOPIC=".nrGraph.myTopic|(:NrSoundRecording)<-[:REPRESENTS {lastModifiedAt, lastModifiedBy}]-(:GlobalSoundRecording)"` -- Select all changes on :REPRESENTS relationships with start nodes of label :NrSoundRecording and end nodes of label :GlobalSoundRecording and only include lastModifiedAt and lastModifiedBy properties in the change event
- `TOPIC_MAPPING_MY_TOPIC=".nrGraph.myTopic|(:NrSoundRecording)<-[:REPRESENTS {-id}]-(:GlobalSoundRecording)"` -- Select all changes on :REPRESENTS relationships with start nodes of label :NrSoundRecording and end nodes of label :GlobalSoundRecording and exclude id property in the change event
- `TOPIC_MAPPING_MY_TOPIC=".nrGraph.myTopic|(:NrSoundRecording)<-[:REPRESENTS {team: 'NR'}]-(:GlobalSoundRecording)"` -- Select all changes on :REPRESENTS relationships identified by key property team equals NR and start nodes of label :NrSoundRecording and end nodes of label :GlobalSoundRecording
- `TOPIC_MAPPING_MY_TOPIC=".nrGraph.myTopic|(:NrSoundRecording)<-[]-()"` -- Select all changes on all relationships that start with nodes of label :NrSoundRecording.
- `TOPIC_MAPPING_MY_TOPIC=".nrGraph.myTopic|(:NrContributor)-[]->(:NrContribution)"` -- Select all changes on all relationships between :NrContributor and :NrContribution nodes. So all 3: IS_MAIN_PERFORMER | IS_FEATURING_PERFORMER | IS_SESSION_MUSICIAN relationships will write to same topic.


### Overriding CDC strategy settings

You can pass a JSON string via the `OVERRIDE_CONFIGS` environment variable to set or override any `neo4j.cdc.*` settings without modifying `config.template.json`. You can check the [reference configuration documentation](https://neo4j.com/docs/kafka/current/source/configuration/#_cdc_strategy_settings):

- Keys already present in the config are replaced with the override value.
- Keys not already present are added to the config.
- Only keys prefixed with `neo4j.cdc.` are accepted — any other keys in the JSON are silently ignored.
- Set to an empty string or `{}` (the default) to disable.

**Locally**, edit `config_override.json` in this folder and `docker-compose up` will pick it up automatically — no need to write a one-liner in `.env`:
```json
{
    "neo4j.cdc.topic.my-topic.key-strategy": "ENTITY_KEYS",
    "neo4j.cdc.topic.my-topic.patterns.0.pattern": "(:OrchardSoundRecording)",
    "neo4j.cdc.topic.my-topic.patterns.0.operation": "update",
    "neo4j.cdc.topic.my-topic.patterns.0.changesTo": "lastModifiedAt",
    "neo4j.cdc.topic.my-topic.patterns.0.txMetadata.executingUser": "my-service-username"
}

```

**In QA/PROD**, set the `OVERRIDE_CONFIGS` environment variable directly with the JSON string:
```
OVERRIDE_CONFIGS='{"neo4j.cdc.poll-interval": "500ms"}'
```