# Neo4j CDC Source (for Neo4j V5+ only)

The Neo4j Source CDC Connector pulls data from Neo4j using CDC strategy, and writes it into Kafka topics.
It uses the database’s Change Data Capture feature to track all changes happening to the database in real time. With CDC, you can identify and respond to
changes (create, update, and delete) on nodes and relationships as they happen, and writes it into Kafka topics so they can be integrate into other systems.

Neo4j extracts CDC information from the transaction log. However, by default the transaction log does not contain information directly usable by CDC.
For CDC to work, the transaction log need to be enriched with further information. This is applied as an extra configuration option to each database.
As soon as CDC is enabled, the database is ready to answer CDC queries from client applications. See Prerequisites on how to do it.

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/source/).


## Prerequisites

- **Enable CDC feature on Neo4j**: Run cmd `ALTER DATABASE graph.db SET OPTION txLogEnrichment "FULL";` to get full enriched messages. More details can be [found here](https://neo4j.com/docs/cdc/current/get-started/self-managed/).
- **Neo4j credentials**: A set of valid credentials (username and password) for a Neo4j database are needed. This user also needs a role that has permission to read CDC logs.
```
// grant CDC access to role that is assigned to all kc_ users. In this we assume that they have architect role
GRANT ACCESS ON DATABASE graph.db TO architect;
GRANT EXECUTE PROCEDURE db.cdc.query ON DBMS TO architect;
GRANT EXECUTE BOOSTED PROCEDURE db.cdc.query ON DBMS TO architect;
```


## 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_ENFORCE_SCHEMA**       | Yes      | `bool`| ` `     | Apply a schema to each record.                                                 |
| **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 change events to publish for each poll cycle.                     |
| **NEO4J_POLL_INTERVAL**           | Yes | `number` | ` `  | The interval in which the database will be polled for changes during 5s suration (valid units are: ms, s, m, h and d; default unit is s).|
| **NEO4J_IGNORE_STORED_OFFSET**           | Yes | `number` | ` `  | Whether to ignore any offset value retrieved from the offset storage saved by a previous run. This is useful for QA where refresh will make the old offset invalid.|
| **NEO4J_BATCH_SIZE**           | Yes | `number` | ` `  | Maximum number of change events to publish for each poll cycle..                    |
| **TOPIC_MAPPING_xxxx**           | Yes | `str` | ` `  | String in format `topicName|pattern selector`. Example: `"cdc.musicGraphV5.labelSoundRecording|(:LabelSoundRecording)"`.|
| **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_source/neo4j_v5