# JDBC Source Connector

## Description

This type of the Kafka Connector queries data form SQL databases using corresponding JDBC driver.
The official documentation is [here]( https://docs.confluent.io/kafka-connectors/jdbc/current/source-connector/source_config_options.html).


### Database access configuration

For local testing create and fill up `.env` file and run
```
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
```

For the instances deployed to AWS we use Lenses' Secret Provider. This extension allows the connector to get secrets from AWS Secrets Manager.

### Example of generated connector config

```json
{
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:mysql://phys.db.devorch.com:3306/art_relations",
    "config.providers": "aws",
    "config.providers.aws.class": "io.lenses.connect.secrets.providers.AWSSecretProvider",
    "config.providers.aws.param.aws.auth.method": "default",
    "config.providers.aws.param.aws.access.key": "dummy-client-key",
    "config.providers.aws.param.aws.secret.key": "dummy-secret-key",
    "config.providers.aws.param.aws.region": "us-east-1",
    "connection.user": "${aws:dev/kafka_connect_jdbc_source/DATABASE_CREDENTIALS:user}",
    "connection.password": "${aws:dev/kafka_connect_jdbc_source/DATABASE_CREDENTIALS:password}",
    "table.whitelist": "country",
    "catalog.pattern": "art_relations",
    "mode": "bulk",
    "quote.sql.identifiers": "never",
    "poll.interval.ms": 5000,
    "topic.prefix": "jdbc-source-ar-"
 }
```

### Snowflake configuration

The connector works in the same manner with Snowflake as with the other SQL databases. The only difference is the way how the connection is configured. We use SSH keys to [connect to Snowflake](https://www.notion.so/Snowflake-7c88cc17b0034e7db669a88fd2962bab?pvs=4#3aa649311d19425f8a3dd2e84eda8348).

Driver jar file: https://mvnrepository.com/artifact/net.snowflake/snowflake-jdbc/3.19.0

#### Local setup

Before running the connector provide the following environment variables in the `.env` file:

```shell
export SNOWFLAKE_CONFIGURATION=true
export SNOWFLAKE_SUBDOMAIN=<your_subdomain>
export SNOWFLAKE_WAREHOUSE=<your_subdomain
export SNOWFLAKE_ROLE=<your_role>
export SNOWFLAKE_DATABASE=<your_database>
export SNOWFLAKE_SCHEMA=<your_schema>
export SNOWFLAKE_USER=<your_user>
export SNOWFLAKE_PRIVATE_KEY_FILENAME=rsa_key.p8
export SNOWFLAKE_PRIVATE_KEY_PASSPHRASE=<your_passphrase>
```

Check that the `platform` is set for your local environment in `docker-compose.yml`:

``` yml
services:
  kafka-connect-jdbc-source:
    platform: linux/arm64 # for Apple Silicon (M1+)
    # or
    platform: linux/amd64 # for Intel Mac
    # etc
```

Check that the Environment is set to local in `docker-compose.yml`

``` yml
environment:
      Environment: 'local'
```

 and `Dockerfile`:

``` Dockerfile
ENV Environment='local'
```

By default, the keys are taken from this path `${HOME}/.ssh/snowflake` you can adjust it in the `docker-compose.yml` file if needed.

#### QA/Prod setup

The following Snowflake config values should be provided as environment variables to the connector instance:

```shell
SNOWFLAKE_CONFIGURATION=true
SNOWFLAKE_SUBDOMAIN=<your_subdomain>
SNOWFLAKE_WAREHOUSE=<your_subdomain
SNOWFLAKE_ROLE=<your_role>
SNOWFLAKE_DATABASE=<your_database>
SNOWFLAKE_SCHEMA=<your_schema>
SNOWFLAKE_USER
```

The private key and its passphrase should be stored in AWS Secrets Manager. The name of the secret must be composed using this pattern:
`{Environment}/${SERVICE_NAME}/SNOWFLAKE_PRIVATE_KEY`
and the secret should have the following format:

```json
{
  "private_key_b64":"<my_base64_private_key>",
  "private_key_passphrase":"<my_passphrase>"
}
```

please note that the private key should be base64 encoded. For initial encoding you can use the following command:

```shell
base64 -i rsa_key.p8
```

#### Snowflake password-Encrypted Private Key support
The Snowflake driver needs the Bouncy Castle cryptography libraries to be able to decrypt private key files protected by a password:
https://docs.snowflake.com/en/user-guide/kafka-connector-install#installing-the-connector-for-confluent

#### Extracting nested object

If you want to extract a field that is originally compose as nested object by a Snowflake SQL query, you should additionally provide the following environment variable:

```shell
EXTRACT_NESTED_OBJECT=true
FIELD_NAME_FOR_KEY=<MY_FIELD_NAME_FOR_KEY>
FIELD_NAME_FOR_VALUE=<MY_FIELD_NAME_FOR_VALUE>
```

For example, if you have a query like this:

```sql
SELECT
    ID,
    OBJECT_CONSTRUCT(
        'id', id,
        'last_modified_at', last_modified_at,
        'nestedField', OBJECT_CONSTRUCT('upc', upc )
    ) AS RECORD,
    LAST_MODIFIED_AT
FROM FACTS.PROD.ORCHARD_PRODUCT;
```

You should provide the following values:

```shell
EXTRACT_NESTED_OBJECT=true
FIELD_NAME_FOR_KEY=ID
FIELD_NAME_FOR_VALUE=RECORD
```

#### Terraform example

Here is an example for Snowflake with a nested object extraction.

<https://github.com/theorchard/terraform-infra/tree/master/dev/kafka-infra/jdbc_source/snowflake_example>