# Neo4j Source Connector

## Description

This type of the Kafka Connector pulls data from Neo4j and writes it into Kafka topics.
This connector is used for workloads where CDC can't be used and instead POLL by query is preferred.

- https://github.com/neo4j-contrib/neo4j-streams/releases

### Neo4j access configuration

For local testing `NEO4J_USERNAME` and `NEO4J_PASSWORD` can be set in `docker-compose.yml` file or in `.env` file.

For the instances deployed to AWS we use Lenses' Secret Provider. This extension allows the connector to get secrets from AWS Secrets Manager. Secret values are used for the following config params:

- neo4j.authentication.basic.username
- neo4j.authentication.basic.password

the values are provided with this format `${aws:<SECRET_NAME>:<SECRET_KEY>}` where `<SECRET_NAME>` is composed of `"<Environment>/<SERVICE_NAME>/credentials"` env variables and `<SECRET_KEY>` is `username` or `password`.

Also, for access AWS secrets manager from local machine please provide temp credentials via `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, ` AWS_SESSION_TOKEN` env variables.

### Connector flags
The following environment variables can be set to change the configuration of the connector:

- `EXTRACT_VALUE_TO_KEY`: Set with the name of a message field to automatically put the value of that field as the key of the Kafka message.

- `DROP_MESSAGE`: Set **true** to set every message's body as `null`. It requires `EXTRACT_VALUE_TO_KEY` to be configured.

- `TRACKING_PROPERTY_NAME`: By default the connector will store internally a timestamp to record the last value it was checked against. To set one from a message field use this property to use as explained in the documentation for [neo4j.query.streaming-property](https://neo4j.com/docs/kafka/5.1/source/configuration/#_query_strategy_settings).

- `NEO4J_QUERY`: Query for sourcing Neo4j data with the connector.

- `NEO4J_DATABASE_NAME`: Name of the Neo4j database to query.


### Running it locally

To Start the server:

```
cp .env.shadow .env

# add your credentials for NEO4J_USERNAME & NEO4J_PASSWORD in .env file
# also make sure AWS_SECRETS_MANAGER=false

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
```


To Trigger an update (if you are connected to dev Aura DB):

```
MERGE (n:TestSource {id: "test-uuid"}) set n.`_timestamp` = timestamp(), n.name = "foo1", n.uuid = apoc.create.uuid() RETURN n;

```


### Key/Value converters

By default `org.apache.kafka.connect.json.JsonConverter` converter is used for both key and value kafka message. Also, schema converter is disabled (which means that connector doesn't expect to have JSON scema in the keys and messages, so basically it can work with arbitrary json format). If you have different type of values provide appropriate converter class.