# Neo4j Sink Connector - V5.1.0 RC2

## Description

The Neo4j Connector for Kafka is designed for customers who want to stream data from Apache Kafka topics into Neo4j or Aura. The connector is compatible with Neo4j 4.4.x and 5.x and Aura 4 and 5. The connector is designed to be compatible with versions of Apache Kafka Connect 2.8 and later. Sink instance which will listen for messages on cofigured topics, and execute a Cypher statement to apply the corresponding change in Neo4j when the messages are received. We are usign Cypher Strategy in this connector.

[Documentataion](https://neo4j.com/docs/kafka/5.1-preview/sink/cypher/).



## Local configuration

The easiest way to test the connector locally is to use dev Kafka cluster and NR Aura. Dev Kafka cluster is available under VPN connection.

You need to copy `.env.shadow` to `.env` and set `NEO4J_USERNAME` and `NEO4J_PASSWORD` values in `.env` file.

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_sink connector successfully updated
neo4j_cdc_sink 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_sink/tasks/0/status
- http://localhost:8083/connectors/neo4j_cdc_sink/config


### Connector configuration options:
Option | Environment variable | Desc
--- | ---| ---
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.authentication.basic.username | NEO4J_USERNAME | Username to authenticate with. Required when neo4j.authentication.type is BASIC.
neo4j.authentication.basic.password | NEO4J_PASSWORD | Password to authenticate with. Required when neo4j.authentication.type is BASIC.


## 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"}
```


## Cypher Queries mapping:


In this case we environmet variable prefixed by `TOPIC_MAPPING_`. Value contains topic name pipe seperated with cypher query.

Format: TOPIC_MAPPING_xxx = "topic.name.caseSensitive|cypher-query-without-newline"

Example:

```
export TOPIC_MAPPING_IDENTITY="cdc.musicgraph.identity|WITH event MERGE (i:Test {id: event.id}) SET i.name = event.name, i.key = event.id, i.lastUpdated = datetime()"
```

**Note:**
- Cyphers should not have a return statement.
- Any double quotes in cypher should be escaped eg: ` i.lastModifiedBy = \"kafka-connect/test_node\",` or use single quotes eg: ` i.lastModifiedBy = 'kafka-connect/test_node',`


Under the hood the connector will create a batch of changes for each topic, and will execute the query with a prepended UNWIND clause. For the above example, the executed query for messages received from `cdc.musicgraph.identity` topic would look like:
```
UNWIND $events AS event
WITH event
MERGE (i:Test {id: event.id})
SET i.name = event.name,
    i.key = event.id,
    i.lastUpdated = datetime()
```

## Use CDC Schema sub-strategy (instead of cypher):

In this case we set the environmet variable `SYNC_SCHEMA_TOPICS`. Value contains comma seperated topic names.

Example:

```
export SYNC_SCHEMA_TOPICS="cdc.musicGraph.product,cdc.artRelations.track,cdc.musicGraph.labelSoundRecording,cdc.musicGraph.isPlacementOf"
```



## Use CDC SOURCE_ID sub-strategy (instead of cypher):

In this case we set the environmet variable `SYNC_SOURCE_ID_TOPICS`. Value contains comma seperated topic names. 
Also set `CHANGE_OPERATION` to true to convert all update operations to created. This is a [feature requested](https://github.com/neo4j/neo4j-kafka-connector/issues/245) from neo4j but SMT is a workaround it. 

Example:

```
export SYNC_SOURCE_ID_TOPICS="cdc.musicGraph.product,cdc.artRelations.track,cdc.musicGraph.labelSoundRecording,cdc.musicGraph.isPlacementOf"
export CHANGE_OPERATION="true"
```