# JDBC Source The JDBC Source Connector enables us to import data from any relational database with an available JDBC driver into Kafka topics. This connector is maintained by Confluent, and deployed as a Kafka Connect plugin. It's official documentation can be [found here](https://docs.confluent.io/kafka-connectors/jdbc/current/source-connector/overview.html). Using the JDBC Source, and a supported JDBC driver, data is loaded by periodically executing a SQL query and creating an output record for each row in the result set. Kafka Connect will track the latest record retrieved so it can continue pulling data from only updated rows if desired. Alternatively, it pulls all data from the source with each execution. ## Features - Copying data from tables - Adding/Removing tables dynamically - Table whitelists and blacklists - Varying polling intervals - Incremental query modes: different modes for tracking row changes depending on a column ## Currently supported Sources As part of our own implementation of the connector, at the moment of writing we have implemented the following drivers: - MySQL JDBC - Snowflake JDBC ## Prerequisites - **Database credentials**: A set of valid credentials (username and password for MySQL, username and private key for Snowflake) with enough permissions to read from the required table(s). ## 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. | | **BATCH_MAX_ROWS** | No | `int` | `100` | Maximum number of rows to include in a single batch when polling for new data. | | **DATABASE_CONNECTION_URL** | Yes | `str` | ` ` | Database URL using to connect to. | | **DATABASE_CATALOG_PATTERN** | No | `str` | `null` | Catalog pattern to fetch table metadata from the DB. Requires `DATABASE_TABLE_WHITELIST` to be set.| | **DATABASE_TABLE_WHITELIST** | No | `str` | ` ` | List of tables to include in copying. Requires `DATABASE_CATALOG_PATTERN` to be set.| | **DATABASE_TABLE_WHITELIST** | No | `str` | ` ` | List of tables to include in copying. Requires `DATABASE_CATALOG_PATTERN` to be set.| | **DATABASE_USER** | No | `str` | ` ` | Database user to connect. Only set for MySQL when `AWS_SECRETS_MANAGER` is false. | | **DATABASE_PASSWORD** | No | `str` | ` ` | Database password to connect. Only set for MySQL when `AWS_SECRETS_MANAGER` is false.| | **DEBUG_MODE** | Yes | `bool`| ` ` | Enable extra debug logging on the connector. | | **EXTRACT_NESTED_OBJECT** | No | `bool`| ` ` | Enable to extract a field part of the message to the key, or a field thats a nested object.| | **FIELD_NAME_FOR_KEY** | No | `str` | ` ` | Name of the field in the message that needs to be extracted to the message's Key. Requires `EXTRACT_NESTED_OBJECT` to be enabled.| | **FIELD_NAME_FOR_VALUE** | No | `str` | ` ` | Name of the nested field in the message that needs to be extracted to the message's value. Requires `EXTRACT_NESTED_OBJECT` to be enabled.| | **INCREMENTING_COLUMN_NAME** | No | `str` | ` ` | Name of the incrementing column used to detect new rows. Empty value indicates it should be autodetected. Only for `incrementing` or `timestmap+incrementing` MODE.| | **MODE** | Yes | `str` | ` ` | The mode for getting updates. Can take `bulk`, `timestamp`, `incrementing`, `timestamp+incrementing`. | | **POLL_INTERVAL_MS** | No | `int` | `5000` | The frequency in ms to poll for new data in each table. | | **TIMESTAMP_COLUMN_NAME** | No | `str` | ` ` | Comma-separated list of timestamp columns to detect new or modified rows. Only for `timestamp` or `timestamp+incrementing` MODE.| | **TIMESTAMP_INITIAL** | No | `str` | `null` | Epoch timestamp (in ms) used for initial queries on `timestamp` OR `timestamp+incrementing` MODE. Use `-1` to use the current time, or empty to retrieve ALL data.| | **TOPIC_PREFIX** | Yes | `str` | ` ` | The name or prefix for the Kafka topic that the connector will produce messages to. | | **QUERY** | No | `str` | ` ` | If specified, the query used by the connector to select new or updated rows. | | **QUERY_SUFFIX** | No | `str` | ` ` | Suffix to append at the end of the generated query. Requires `QUERY` to be set. | | **QUOTE_SQL_IDENTIFIERS** | No | `str` | `always`| By default `always` is set, which will quote table/column names and other SQL identifiers.| | **VALIDATE_NON_NULL** | No | `bool`| `true` | The connector will always validate that incrementing/timestamp tables have NOT NULL set for the columns. If the tables don't, set this to `false`.| ### Snowflake configuration If you are setting up a Snowflake Source, you will have to set up a user with private key for Auth, and use this configuration properties: | Environment Variable | Required | Type | Default | Description | |------------------------------- |--------- |------ |-------- |------------------------------------------------------------------------------------ | | **SNOWFLAKE_CONFIGURATION** | Yes | `bool`| ` ` | Enable to indicate the connector will use Snowflake private key auth. | | **SNOWFLAKE_DATABASE** | Yes | `str` | ` ` | Snowflake Database to use to run the connector's queries on. | | **SNOWFLAKE_ROLE** | Yes | `str` | ` ` | Snowflake Role to use with the user to run the connector's queries. | | **SNOWFLAKE_SCHEMA** | Yes | `str` | ` ` | Snowflake Schema to use to run the connector's queries on. | | **SNOWFLAKE_SUBDOMAIN** | Yes | `str` | ` ` | Snowflake instance Subdomain. | | **SNOWFLAKE_USER** | Yes | `str` | ` ` | Snowflake User to use to run the connector's queries. | | **SNOWFLAKE_WAREHOUSE** | Yes | `str` | ` ` | Snowflake Warehouse to use to run the connector's queries. | ## 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}/DATABASE_CREDENTIALS` **Secret's Keys**: - **username**: The DB Username. - **password**: The DB User's password. To access Secrets Manager from a local instance, remember to set AWS credentials on your environment variables. ### Snowflake Private key Secrets The service user for the Snowflake-JDBC connector requires to be set up with key pair authentication [Guide](https://www.notion.so/Snowflake-7c88cc17b0034e7db669a88fd2962bab#3aa649311d19425f8a3dd2e84eda8348). Once the user is set up with the key, you will need to create a secret for the private key and it's passphrase. **Secret's Name Structure**: `${Environment}/${SERVICE_NAME}/SNOWFLAKE_PRIVATE_KEY` **Secret's Keys**: - **private_key_b64**: The private key encoded in base64. - **private_key_passphrase**: The encrypted private key passphrase. To correctly decrypt the private key, the Snowflake JDBC driver requires the use of BouncyCastle libraries, which are included as part of our implementation of the connector. - https://docs.snowflake.com/en/user-guide/kafka-connector-install#installing-the-connector-for-confluent - https://community.snowflake.com/s/article/Key-pair-authentication-failure-due-to-ObjectIdentifier-or-SecretKeyFactory-not-available-errors ## Extracting message keys ### 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, set the ENV var `EXTRACT_NESTED_OBJECT` to `true` in conjuction with `FIELD_NAME_FOR_KEY` with the name of the field from which you want to extract 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. ### Extracting nested objects For some cases, like Snowflake queries returning nested JSON-like objects, you can extract the value. Set the ENV var `EXTRACT_NESTED_OBJECT` to `true` in conjuction with `FIELD_NAME_FOR_VALUE` to indicate the field in the message containing the object. ## Terraform Infra As a recommendation, instead of setting the SQL query directly as an environment variable, you can write it on a `.sql` file and pass it to the ENV in Terraform like this: ```HCL replace(templatefile("${path.module}/query.sql", {}), "\n", " ") ``` :::warning Memory Considerations In some instances the connector's container might require a bigger memory allocation than normal. We have found that Snowflake-JDBC connectors could require a minimum of **3GB of memory** to work steadily; with less than that the connector might enter a crash loop. ::: ### Existing examples - [JDBC Source in Terraform Infra](https://github.com/theorchard/terraform-infra/tree/master/qa/kafka-infra/jdbc_source)