# Google Sheets Stream Source Connector

## Description

This Kafka Connector is used to poll for changes in Google Sheets.

### Connector configuration

For local testing following environment variables should be set in `.env` file.

`GCP_APP_NAME` - Google sheets application name, 
defined under ***API & Services** -> [OAuth Consent Screen](https://console.cloud.google.com/apis/credentials/consent)*. 
When created application scope should be set to `https://www.googleapis.com/auth/spreadsheets.readonly`.

`SPREADSHEET_ID` - ID is copied from spreadsheet URL `https://docs.google.com/spreadsheets/d/<SPREADSHEET ID>/`

`SPREADSHEET_RANGE` - The range can be optionally defined as spreadsheet name or spreadsheet name + cell range (i.e. `dev!A1:D5`) 

`SPREADSHEET_READ_DIMENSION` - The dimention in which data is being read, it could be row by row or column by column, the default is `ROWS`

`CONNECTOR_SPLIT_RESULTS` - This option force the connector to split read rows/columns into separate messages. Connector reads the whole range anyways.

### Secrets configuration

Connector credentials are set in two separate menus of the GCP. 

1. Create **client Id** and **client Secret** under 
***API & Services** -> [Credentilas](https://console.cloud.google.com/apis/credentials)*. 
This connector uses OAuth2 credentials for web applications. 

2. Using generated **client Id** and **client Secret** create **accessToken** and **refreshToken** in 
*[OAuth 2.0 Playground](https://developers.google.com/oauthplayground/)*. 
Next choose `Use your own OAuth credentials` in `OAuth 2.0 configuration`, select 
the `https://www.googleapis.com/auth/spreadsheets.readonly` scope and click `Authorize API` button. 
Once done click the `Exchange authorization code for tokens` button, you will get a refresh and an access token 
which is required to access OAuth protected resources.


https://console.cloud.google.com/apis/credentials
To access *Google Cloud Platform* credentials stored in *AWS SecretsManager* we use *Lenses' Secret Provider*. 
Secret values are used for the following config params:

```
camel.source.endpoint.clientId
camel.source.endpoint.clientSecret
camel.source.endpoint.accessToken
camel.source.endpoint.refreshToken
```

The values are vere it is required to retrieve a secret are provided in this format `${aws:<AWS_SECRETS_MANAGER_KEY>:<SECRET_KEY>}` where `<AWS_SECRETS_MANAGER_KEY>` is a secret name in AWS SecretsManager and `<SECRET_KEY>` is a key from key/value pair of the secret. `AWS_REGION` should be set to relevant region used in AWS

To access AWS secrets manager from local machine please generate temporary credentials using [aws-creds-generator](https://github.com/theorchard/collab/tree/master/jcarrion/aws-creds-generator). No additional actions required.

### Message size (a.k.a. max cell range)

Seems like we can do 1,799,820 cells (18 cols x 99 990 rows) for ~15 MB messages in kafka (this is current message size for most of the 
topics per our current terraform config) assuming we have ~6.5 average chars.

### Key/Value converters

By default `org.apache.kafka.connect.storage.StringConverter` are used for keys and values in kafka messages respectively. Also, schema converter is disabled (`value.converter.schemas.enable`).

### Convert JSON to AVRO messages

This connector doesn't support confluence AVRO converter (`io.confluent.connect.avro.AvroConverter`). 
When table data is polled by connector in single message (splitResults=false) it is represented by nested 
arrays of strings.To convert JSONS messages to AVRO it is required to generate two streams and convert 
messages on the fly.
See the example:

```sql
CREATE STREAM source_json_stream (
    majorDimension VARCHAR, range VARCHAR, "VALUES" ARRAY<ARRAY<string>>
) WITH (
    KAFKA_TOPIC='source.json.kafka.topic',
    VALUE_FORMAT='JSON'
);

CREATE STREAM target_avro_stream
WITH (
    VALUE_FORMAT='AVRO',
    KAFKA_TOPIC='target.avro.kafka.topic'
  )
AS SELECT
    "VALUES"[1] AS firstTopicField,
    "VALUES"[2] AS secondTopicField
FROM source_json_stream;
```
*Please note*: The elements of an KSQL array are one-indexed. For example, `SOME_ARRAY[1]` retrieves the first element from the array.

### Example of generated connector config

To generate connector configuration (by default stored in `config.json`) we use `config.py` python script 
which takes shell env variables and adds them to `config.json.jinja` configuration template
```json
{
  "name": "ibolshakov_gsheets_test.cols.local",
  "connector.class": "org.apache.camel.kafkaconnector.googlesheetsstream.CamelGooglesheetsstreamSourceConnector",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter.schemas.enable": "false",
  "topics": "ibolshakov_gsheets_test_cols",
  "camel.source.path.apiName": "DATA",
  "camel.source.endpoint.applicationName": "kafka-gsheets-source-dev",
  "camel.source.endpoint.delay": "60000",
  "camel.source.endpoint.scopes": "https://www.googleapis.com/auth/spreadsheets.readonly",
  "camel.source.endpoint.spreadsheetId": "14stMqaqyJL0aJayfX1jY3g6ejHBcQcg4sElJfexrPg4",
  "camel.source.endpoint.valueRenderOption": "UNFORMATTED_VALUE",
  "camel.source.endpoint.range": "Roster!A2:R9",
  "camel.source.endpoint.majorDimension": "COLUMNS",
  "camel.source.endpoint.runLoggingLevel": "DEBUG",
  "camel.source.endpoint.clientId": "${aws:dev/kafka-connect-gsheets-source/google_api_credentials:GOOGLE_CLIENT_ID}",
  "camel.source.endpoint.clientSecret": "${aws:dev/kafka-connect-gsheets-source/google_api_credentials:GOOGLE_CLIENT_SECRET}",
  "camel.source.endpoint.accessToken": "${aws:dev/kafka-connect-gsheets-source/google_api_credentials:GOOGLE_OAUTH2_ACCESS_TOKEN}",
  "camel.source.endpoint.refreshToken": "${aws:dev/kafka-connect-gsheets-source/google_api_credentials:GOOGLE_OAUTH2_REFRESH_TOKEN}",
  "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"
}
```

If you wish to output 1 kafka message per row, you will need to set the following config
- `CONNECTOR_SPLIT_RESULTS=true`
- `VALUE_RENDER_OPTION=FORMATTED`
- `CONNECT_VALUE_CONVERTER='org.apache.kafka.connect.json.JsonConverter'`