# Neo4j CDC Sink (for Neo4j V5+ only)

The Neo4j CDC Sink Connector consumes messages from Apache Kafka topics and applies configured changes into a Neo4j or Aura database.

The Sink connector can use different strategies to write Kafka messages into a Neo4j database:
- Cypher: Most commonly use by us. Use the incoming Kafka messages as parameters of a user-provided Cypher write query.
- Change Data Capture: Process incoming Kafka messages generated by either a Source instance configured with Change Data Capture strategy or the deprecated Neo4j Streams plugin, then apply the changes directly to the target database.
More details can be [found here](https://neo4j.com/docs/kafka/5.1/sink/).

This connector is maintained and supported by Neo4j, and deployed as a Kafka Connect plugin. It's official documentation for version 5.1.11 can be [found here](https://neo4j.com/docs/kafka/5.1/sink/).


## Prerequisites

- **Neo4j credentials**: A set of valid credentials (username and password) for a Neo4j database are needed.


## Configuration Properties

The following are the configuration properties specific to this connector that are exposed through Environment Variables in the Connector instance.
Note that common environment variables are described in the [Deploying Connectors doc](/docs/kafka-connect/deploying-connectors.mdx).

| Environment Variable           | Required | Type  | Default | Description                                                                    |
|------------------------------- |--------- |------ |-------- |------------------------------------------------------------------------------- |
| **AWS_SECRETS_MANAGER**        | No       | `bool`| `false`  | Enable to use AWS Secrets Manager for credentials, instead of ENV variables: NEO4J_USERNAME & NEO4J_PASSWORD.   |
| **DEBUG_MODE**                 | No       | `bool`| ` `      | Enable extra debug logging on the connector.                                   |
| **NEO4J_DATABASE_NAME**        | No       | `str`| `graph.db`| Name of the database to connect to.                                           |
| **NEO4J_KEY_CONVERTER**        | Yes      | `str` | ` `     | Kafka converter class for message keys.                                        |
| **NEO4J_KEY_CONVERTER_SCHEMAS_ENABLE** | Yes | `str` | ` `  | Enable when JSON messages include the schema.                                  |
| **NEO4J_VALUE_CONVERTER**      | Yes      | `str` | ` `     | Kafka converter class for message values.                                      |
| **NEO4J_VALUE_CONVERTER_SCHEMAS_ENABLE** | Yes | `str` | ` `  | Enable when JSON messages include the schema.                                |
| **NEO4J_BATCH_SIZE**           | Yes | `number` | ` `  | Maximum number of messages processed per transaction per topic.                     |
| **NEO4J_BATCH_EXEC_TIMEOUT**   | Yes | `number` | ` `  | Maximum amount of time a batch is allowed to be processed (valid units are: ms, s, m, h and d; default unit is s).|
| **SYNC_SCHEMA_TOPICS**        | No | `str` | ` `  | Comma seperated list of topicNames. This is to turn ON CDC [Schema sub-strategy](https://neo4j.com/docs/kafka/5.1/sink/cdc/#schema).|
| **SYNC_SOURCE_ID_TOPICS**     | No | `str` | ` `  | If SYNC_SCHEMA_TOPICS is not TRUE then, this is Comma seperated list of topicNames to sync usong CDC [Source ID sub-strategy](https://neo4j.com/docs/kafka/5.1/sink/cdc/#source-id).|
| **TOPIC_MAPPING_xxxx**           | Yes | `str` | ` `  | String in format `topicName|cypher without new lines`. Example: `"event.snowflake.newmusicfriday.newtrack|${local.newmusicfriday_newtrack}"`. This used [Cypher sink](https://neo4j.com/docs/kafka/5.1/sink/cypher/) strategy. |
| **NEO4J_SERVER_URI**           | Yes      | `str` | ` `     | Neo4j cluster URI starting with neo4j+s://.                                    |
| **NEO4J_USERNAME**             | No[^1]   | `str` | ` `     | Neo4j username. Only with `AWS_SECRETS_MANAGER` set to false.                  |
| **NEO4J_PASSWORD**             | No[^1]   | `str` | ` `     | Neo4j password. Only with `AWS_SECRETS_MANAGER` set to false.                  |
| **NEO4J_START_FROM**             | No[^1]   | `str` | `NOW`     | A time anchor to start streaming from. 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.|



## Secrets

This connector expects to find the following secret in AWS Secrets Manager when `AWS_SECRETS_MANAGER` is **enabled**.

**Secret's Name Structure**: `${Environment}/${SERVICE_NAME}/NEO4J_CREDENTIALS`

**Secret's Keys**:
- **username**: The Neo4j Username.
- **password**: The Neo4j User's password.

To access Secrets Manager from a local instance, remember to set AWS credentials on your environment variables using `awsume dev`.


## Terraform Infra

Refer https://github.com/theorchard/terraform-infra/tree/master/qa/kafka-infra/neo4j_cdc_sink