# Neo4j Source (Cypher)

The Neo4j Source Connector pulls data from Neo4j by running a Cypher query, and writes it into Kafka topics.

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

- **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**        | Yes      | `bool`| ` `     | Enable to use AWS Secrets Manager for credentials, instead of ENV variables.   |
| **DEBUG_MODE**                 | Yes      | `bool`| ` `     | Enable extra debug logging on the connector.                                   |
| **DROP_MESSAGE**               | No       | `bool`| ` `     | Enable to set every message's body as `null` (tombstone). Requires `EXTRACT_VALUE_TO_KEY` to be set.|
| **EXTRACT_VALUE_TO_KEY**       | No       | `str` | ` `     | Extract a message property's value and set it as the message key.              |
| **LAST_CHECK_PROP**            | No       | `str` | ` `     | Message property name to update last_check bookmark, to use instead of connector's internal default timestamp.|
| **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_QUERY**                | Yes      | `str` | ` `     | Cypher query to poll data from Neo4j.                                          |
| **NEO4J_SERVER_URI**           | Yes      | `str` | ` `     | Neo4j cluster URI.                                                             |
| **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.                  |
 

[^1]: Only required when Secrets Manager is not used / AWS_SECRETS_MANAGER is set to false.

## 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.

## Extracting message keys, and tombstone records

### Message Value to Key extraction

For some pipelines you may desire to automatically make the connector set the Kafka message key by extracting the value from a field present in the message's body. To enable this, use the ENV var `EXTRACT_VALUE_TO_KEY`, specifying the name of the field you want as the key.

This is possible through an implementation of Kafka's Simple Message Transforms (SMTs). Using a combination of [ExtractField$Key](https://docs.confluent.io/platform/current/connect/transforms/extractfield.html) with [ValueToKey](https://docs.confluent.io/platform/current/connect/transforms/valuetokey.html), which allows the connector to automatically perform the operation on every message before pushing it to the target topic.

### Tombstone records

A tombstone record is defined as a message with a unique key and a `null` body, often used to indicate a deletion operation in downstream systems.

To implement this in the connector, first configure it with `EXTRACT_VALUE_TO_KEY` to specify a field (usually an ID) to be set as key, and then enable `DROP_MESSAGE`, which will set every message's body to be `null`.

## Terraform Infra

As a recommendation, instead of setting the Cypher query directly as an environment variable, you can write it on a `.cypher` file and pass it to the ENV in Terraform like this:

```HCL
replace(file("my_query.cypher"), "\n", " ")
```