# Snowflake Source Connector for Confluent Cloud The fully managed Snowflake Source connector for Confluent Cloud can capture a snapshot of the existing data in specified Snowflake tables and then monitor and record all subsequent row-level changes to that data. The connector supports AVRO, JSON Schema, and PROTOBUF output data formats. All of the events for each table are recorded in a separate Apache Kafka® topic. The events can then be easily consumed by applications and services. Note that deleted records are not captured. #### NOTE - If you require private networking for fully managed connectors, make sure to set up the proper networking beforehand. For more information, see [Manage Networking for Confluent Cloud Connectors](../networking/internet-resource.md#clusters-connect-cloud). ## Features The Snowflake Source connector provides the following features: * **Topics created automatically**: The connector automatically creates Kafka topics using the naming convention: ``. The topics are created with the properties: `topic.creation.default.partitions=1` and `topic.creation.default.replication.factor=3`. * **Modes**: Set one of the following modes for updating a table each time it is polled: - **bulk**: Performs a bulk load of all eligible table each time it is polled. - **incrementing**: Uses a strictly incrementing column on each table to detect only new rows. Only rows with non-null value of incrementing column will be captured. - **timestamp**: Uses timestamp column(s) to detect new and modified rows. On specifying multiple timestamp columns, the COALESCE SQL function will be used to find out the effective timestamp for a row. This assumes that the effective timestamp is updated with each write and its values are monotonically incrementing, but not necessarily unique. Only rows with non-null value of effective timestamp will be captured. - **timestamp+incrementing**: Uses two columns, a timestamp column that detects new and modified rows, and a strictly incrementing column, which provides a globally unique ID for updates, so each row can be assigned a unique stream offset. Only rows with non-null effective timestamp value and non-null incrementing column value will be captured. * **Database authentication:** The connector supports private key (with or without passphrase) authentication. * **Data formats:** The connector supports AVRO, JSON Schema, and PROTOBUF output data. [Schema Registry](../../get-started/schema-registry.md#cloud-sr-config) must be enabled to use these formats. * **Offset management capabilities**: Supports offset management. For more information, see [Manage custom offsets](#cc-snowflake-source-custom-offsets). * **Client-side encryption (CSFLE and CSPE) support**: The connector supports CSFLE and CSPE for sensitive data. For more information about CSFLE or CSPE setup, see [connector configuration](#cc-snowflake-source-setup-connection). For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](../connect-api-section.md#ccloud-connect-api) section. ## Limitations Be sure to review the following information. * For connector limitations, see [Snowflake Source Connector](../limits.md#cc-snowflake-source-limits) limitations. * If you plan to use one or more Single Message Transformations (SMTs), see [SMT Limitations](../single-message-transforms.md#cc-single-message-transforms-limitations). ## Manage custom offsets You can manage the offsets for this connector. Offsets provide information on the point in the system from which the connector is accessing data. For more information, see [Manage Offsets for Fully Managed Connectors in Confluent Cloud](../offsets.md#connect-custom-offsets). **To manage offsets**: - Manage offsets using Confluent Cloud APIs. For more information, see [Connect offsets API reference](https://docs.confluent.io/cloud/current/ccloud/offsets-connect-v-1/). ### Mode-wise offset structure ### Bulk In bulk mode, since the entire table is queried during each poll, offsets are not used. ### Incrementing Each table has an individual entry in the offsets topic, with the table name stored in the key of the offset message. The connector fetches rows where the value of the incrementing column is strictly greater than the value stored in the offset. **Key** ```bash ["SfConn-202",{"table":"TEST_DB.PUBLIC.TEST_TS_MODE"}] ``` **Value** ```bash { "incrementing": 2 } ``` Value includes the following information: - `incrementing`: Specifies the value of incrementing column till which connector has completed reading. ### Timestamp **Key** ```bash ["SfConn-203",{"table":"TEST_DB.PUBLIC.TEST_TS_MODE_1"}] ``` **Value** ```bash { "timestamp_nanos": 463000000, "timestamp": 1741253351000 } ``` Values include the following information: - `timestamp`: Represents the number of milliseconds since January 1, 1970, 00:00:00 UTC. - `timestamp_nanos`: Represents the fractional seconds component of a Timestamp object. ### Timestamp+Incrementing This mode is a combination of the previous two modes. Hence, its offsets also represent a combination of the two modes. **Key** ```bash ["SfConn-203",{"table":"TEST_DB.PUBLIC.TEST_TS_MODE_1"}] ``` **Value** ```bash { "timestamp_nanos": 352462000, "incrementing": 2, "timestamp": 1713218242000 } ``` ### Mode-wise offset guidance #### Bulk mode There are no offsets in the case of bulk mode since the whole database is queried in each poll method. #### Timestamp mode ### Get the current offset To get the current offset, make a `GET` request that specifies the environment, Kafka cluster, and connector name. ```bash GET /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets Host: https://api.confluent.cloud ``` **Response:** Successful calls return HTTP `200` with a JSON payload that describes the offset. ```bash { "id": "lcc-example123", "name": "{connector_name}", "offsets": [ { "partition": { "table": "{table_name}" }, "offset": { "timestamp": 1741577430000, "timestamp_nanos": 273000000 } } ], "metadata": { "observed_at": "2025-03-10T04:00:31.754227738Z" } } ``` Responses include the following information: - The position of latest offset. - The observed time of the offset in the metadata portion of the payload. The `observed_at` time indicates a snapshot in time for when the API retrieved the offset. A running connector is always updating its offsets. Use `observed_at` to get a sense for the gap between real time and the time at which the request was made. By default, offsets are observed every minute. Calling `GET` repeatedly will fetch more recently observed offsets. - Information about the connector. - In these examples, the curly braces around `connector_name` indicate a replaceable value. ### Update the offset To update the offset, make a `POST` request that specifies the environment, Kafka cluster, and connector name. Include a JSON payload that specifies new offset and a patch type. ```bash POST /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets/request Host: https://api.confluent.cloud { "type": "PATCH", "offsets": [ { "partition": { "table": "{table_name}" }, "offset": { "timestamp": 1741577430000, "timestamp_nanos": 273000000 } } ] } ``` **Considerations:** - You can only make one offset change at a time for a given connector. - This is an asynchronous request. To check the status of this request, you must use the check offset status API. For more information, see **Get the status of an offset request**. - For source connectors, the connector attempts to read from the position defined by the requested offsets. **Response:** Successful calls return HTTP `202 Accepted` with a JSON payload that describes the offset. ```bash { "id": "lcc-example123", "name": "{connector_name}", "offsets": [ { "partition": { "table": "{table_name}" }, "offset": { "timestamp": 1741577430000, "timestamp_nanos": 273000000 } } ], "requested_at": "2025-03-10T04:06:30.786486181Z", "type": "PATCH" } ``` Responses include the following information: - The requested position of the offsets in the source. - The time of the request to update the offset. - Information about the connector. ### Delete the offset To delete the offset, make a `POST` request that specifies the environment, Kafka cluster, and connector name. Include a JSON payload that specifies the delete type. ```bash POST /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets/request Host: https://api.confluent.cloud { "type": "DELETE" } ``` **Considerations:** - Delete requests delete the offset for the provided partition and reset to the base state. A delete request is as if you created a fresh new connector. - This is an asynchronous request. To check the status of this request, you must use the check offset status API. For more information, see **Get the status of an offset request**. - Do not issue delete and patch requests at the same time. - For source connectors, the connector attempts to read from the position defined in the base state. **Response**: Successful calls return HTTP `202 Accepted` with a JSON payload that describes the result. ```bash { "id": "lcc-example123", "name": "{connector_name}", "offsets": [], "requested_at": "2025-03-10T04:11:14.263641766Z", "type": "DELETE" } ``` Responses include the following information: - Empty offsets. - The time of the request to delete the offset. - Information about Kafka cluster and connector. - The type of request. ### Get the status of an offset request To get the status of a previous offset request, make a `GET` request that specifies the environment, Kafka cluster, and connector name. ```bash GET /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets/request/status Host: https://api.confluent.cloud ``` **Considerations:** - The status endpoint always shows the status of the most recent PATCH/DELETE operation. **Response**: Successful calls return HTTP `200` with a JSON payload that describes the result. The following is an example of an applied patch. ```bash { "request": { "id": "lcc-example123", "name": "{connector_name}", "offsets": [ { "partition": { "table": "{table_name}" }, "offset": { "timestamp": 1741577430000, "timestamp_nanos": 273000000 } } ], "requested_at": "2025-03-10T04:06:30.786486181Z", "type": "PATCH" }, "status": { "phase": "APPLIED", "message": "The Connect framework-managed offsets for this connector have been altered successfully. However, if this connector manages offsets externally, they will need to be manually altered in the system that the connector uses." }, "previous_offsets": [ { "partition": { "table": "{table_name}" }, "offset": { "timestamp": 1741577430000, "timestamp_nanos": 273000000 } } ], "applied_at": "2025-03-10T04:06:32.413138178Z" } ``` Responses include the following information: - The original request, including the time it was made. - The status of the request: applied, pending, or failed. - The time you issued the status request. - The previous offsets. These are the offsets that the connector last updated prior to updating the offsets. Use these to try to restore the state of your connector if a patch update causes your connector to fail or to return a connector to its previous state after rolling back. #### Incrementing mode and Incrementing+timestamp mode For **incrementing** and **timestamp+incrementing** mode, the API request payload and response match those for **timestamp** mode. Simply replace the `offset: {}` block with the specified offsets from their respective sections above. ### Offset caveats Note the following important consideration for managing offsets: - **Timestamp mode**: If you need to change the fractional seconds value while resetting offsets, you should do this in `timestamp_nanos`. - **Schema evolution**: Confluent does not recommend updating offsets to a point in time before any DDL changes to the tables, as this may cause inconsistent results. If you move the offsets back to a point before the schema evolution of the targeted tables during the connector run, you will not retrieve the exact same records as before. The records will reflect the schema of the current table. ### JSON payload The table below offers a description of the unique fields in the JSON payload for managing offsets of the Snowflake Source connectors: | Field | Definition | Required/Optional | |-------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------------| | `incrementing` | Specifies the value of incrementing column up to which the connector has
processed. The connector gets only values greater than the value in this field.

Available only in the following modes: incrementing, timestamp+incrementing. | Required | | `table` | The name of the table.

Available in the following modes: incrementing, timestamp, timestamp+incrementing. | Required | | `timestamp` | The number of milliseconds since `January 1, 1970, 00:00:00` GMT represented by the Timestamp object of the column value.

Available only in the following modes: timestamp, timestamp+incrementing. | Required | | `timestamp_nanos` | Fractional seconds component of the Timestamp object.

Available only in the following modes: timestamp, timestamp+incrementing. | Required | ## Generate a Snowflake key pair Before you create the connector, you need to generate a key pair. Snowflake authentication requires 2048-bit (minimum) RSA. You add the public key to a Snowflake user account. You add the private key to the connector configuration (when completing the Quick Start instructions). #### NOTE * This procedure generates an unencrypted private key. You can generate and use an encrypted key. If you generate an encrypted key, add the passphrase to your connector configuration in addition to the private key. For information about generating an encrypted key, see [Using Key Pair Authentication](https://docs.snowflake.com/en/user-guide/kafka-connector-install.html#using-key-pair-authentication-key-rotation) in the Snowflake documentation. * When you use a non-encrypted private key, you might see the following configuration validation error. Check whether your private key is valid or consider using an encrypted private key. ![Private key validation error](images/ccloud-snowflake-key-validation-error.png) ### Creating the key pair Complete the following steps to generate a key pair. 1. Generate a private key using OpenSSL. ```none openssl genrsa -out snowflake_key.pem 2048 ``` 2. Generate the public key referencing the private key. ```none openssl rsa -in snowflake_key.pem -pubout -out snowflake_key.pub ``` 3. List the generated Snowflake key files. ```none ls -l snowflake_key* -rw-r--r-- 1 1679 Jun 8 17:04 snowflake_key.pem -rw-r--r-- 1 451 Jun 8 17:05 snowflake_key.pub ``` 4. Show the contents of the public key file. ```none cat snowflake_key.pub -----BEGIN PUBLIC KEY----- MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA2zIuUb62JmrUAMoME+SX vsz9KUCp/cC+Y+kTGfYB3jRDQ06O0UT+yUKMO/KWuc0dUxZ8s9koW5l/n+TBfxIQ ... omitted 1tD+Ktd/CTXPoVEI2tgCC9Avf/6/9HU3IpV0gL8SZ8U0N5ot4Uw+CSYB3JjMagEG bBWZ8Qc26pFk7Fd17+ykH6rEdLeQ9OElc0ZruVwSsa4AxaZOT+rqCCP7FQPzKTtA JQIDAQAB -----END PUBLIC KEY----- ``` 5. Copy the key, capturing only the portion between `--BEGIN PUBLIC KEY--` and `--END PUBLIC KEY--`. You can do this manually, or by using the following command: ```none grep -v "BEGIN PUBLIC" snowflake_key.pub | grep -v "END PUBLIC"|tr -d '\r\n' ``` You will add this to a new user in Snowflake. In the following section, you create a user and add the public key. ### Creating a user and adding the public key Open your Snowflake project. Complete the following steps to create a user account and add the public key to this account. 1. Go to the **Worksheets** panel and switch to the **SECURITYADMIN** role. #### IMPORTANT Be sure to set the SECURITYADMIN role in the **Worksheets** panel (shown below) and not by using the user account drop-down selection. For additional information, see [User Management](https://docs.snowflake.com/en/user-guide/admin-user-management.html#user-roles). ![Snowflake security admin role](images/ccloud-snowflake-security-admin.png) 2. Run the following query in Worksheets to create a user, and add the public key copied earlier. ```none CREATE USER confluent RSA_PUBLIC_KEY=''; ``` Make sure to add the public key as a **single line** in the statement.The following shows what this looks like in Snowflake Worksheets: ![Snowflake sysadmin role creation statements](images/ccloud-snowflake-query-example.png) ### Configuring user privileges Complete the following steps to set the correct privileges for the user added. For example: Suppose you want to send Apache Kafka® records to a database named `PRODUCTION` using the schema `PUBLIC`. The following shows the required queries to configure the necessary user privileges. ```bash // Use a role that can create and manage roles and privileges: use role securityadmin; // Create a Snowflake role with the privileges to work with the connector create role kafka_connector_role; // Grant privileges on the database: grant usage on database PRODUCTION to role kafka_connector_role; // Grant privileges on the schema: grant usage on schema PRODUCTION.PUBLIC to role kafka_connector_role; grant create table on schema PRODUCTION.PUBLIC to role kafka_connector_role; grant create stage on schema PRODUCTION.PUBLIC to role kafka_connector_role; grant create pipe on schema PRODUCTION.PUBLIC to role kafka_connector_role; // Grant the custom role to an existing user: grant role kafka_connector_role to user confluent; // Make the new role the default role: alter user confluent set default_role=kafka_connector_role; ``` #### NOTE Grant privileges directly to the role to work with the connector. Privileges do not inherit from the role hierarchy. ### Extracting the private key You add the private key to your Snowflake connector configuration. Extract the key and put it in a safe place until you set up your connector. 1. List the generated Snowflake key files. ```none ls -l snowflake_key* -rw-r--r-- 1 1679 Jun 8 17:04 snowflake_key.pem -rw-r--r-- 1 451 Jun 8 17:05 snowflake_key.pub ``` 2. Show the contents of the private key file. ```none cat snowflake_key.pem -----BEGIN RSA PRIVATE KEY----- MIIEpQIBAAKCAQEA2zIuUb62JmrUAMoME+SXvsz9KUCp/cC+Y+kTGfYB3jRDQ06O 0UT+yUKMO/KWuc0dUxZ8s9koW5l/n+TBfxIQx+24C2+l9t3TxxaLdf/YCgQwKNR9 dO9/c+SkX8NfcwUynGEo3wpmdb4hp0X9TfWKX9vG//zK2tndmMUrFY5OcGSSVJYJ Wv3gk04sVxhINo5knpgZoUVztxcRLm/vNvIX1tD+Ktd/CTXPoVEI2tgCC9Avf/6/ 9HU3IpV0gL8SZ8U0N5ot4Uw+CSYB3JjMagEGbBWZ8Qc26pFk7Fd17+ykH6rEdLeQ ... omitted UfrYj7+p03yVflrsB+nyuPETnRJx41b01GrwJk+75v5EIg8U71PQDWfy1qOrUk/d 9u25iaVRzi6DFM0ppE76Lh72SKy+m0iEZIXWbV9q6vf46Oz1PrtffAzyi4pyJbe/ ypQ53f0CgYEA7rE6Dh0tG7EnYfFYrnHLXFC2aVtnkfCMIZX/VIZPX82VGB1mV43G qTDQ/ax1tit6RHDBk7VU4Xn545Tgj1z6agYPvHtkhxYTq50xVBXr/xwlMnzUZ9s3 VjGpMYQANm2seleV6/si54mT4TkUyB7jMgWdFsewtwF60quvxmiA9RU= -----END RSA PRIVATE KEY----- ``` 3. Copy the key. You will add it to the connector configuration. Copy only the part of the key between `--BEGIN RSA PRIVATE KEY--` and `--END RSA PRIVATE KEY--`). You can do this manually or you can use the following command: ```none grep -v "BEGIN RSA PRIVATE KEY" snowflake_key.pem | grep -v "END RSA PRIVATE KEY"|tr -d '\r\n' ``` 4. Save the key to use later when you are completing the Quick Start steps. Or, you can complete the previous step when you actually need to get the key for the connector config. ## Quick Start Use this quick start to get up and running with the Confluent Cloud Snowflake Source connector. The quick start provides the basics of selecting the connector and configuring it to obtain a snapshot of the existing data in specified Snowflake tables and then monitoring and recording all subsequent row-level changes. Prerequisites : - Authorized access to a [Confluent Cloud](https://www.confluent.io/confluent-cloud/) cluster on Amazon Web Services (AWS), Microsoft Azure (Azure), or Google Cloud. - The Confluent CLI installed and configured for the cluster. See [Install the Confluent CLI](https://docs.confluent.io/confluent-cli/current/install.html). - The connector automatically creates Kafka topics using the naming convention: ``. The topics are created with the properties: `topic.creation.default.partitions=1` and `topic.creation.default.replication.factor=3`. If you want to create topics with specific settings, create the topics before running this connector. - [Schema Registry](../../get-started/schema-registry.md#cloud-sr-config) must be enabled. - Make sure your connector can reach your service. Consider the following before running the connector: * Depending on the service environment, certain network access limitations may exist. See [Manage Networking for Confluent Cloud Connectors](../networking/internet-resource.md#clusters-connect-cloud) for details. * To use a set of public egress IP addresses, see [Public Egress IP Addresses for Confluent Cloud Connectors](../static-egress-ip.md#cc-static-egress-ips). For additional fully managed connector networking details, see [Networking and DNS](../overview.md#connect-internet-access-resources). * Clients from Azure Virtual Networks are not allowed to access the server by default. Check that your Azure Virtual Network is correctly configured and that **Allow access to Azure Services** is enabled. * See your specific cloud platform documentation for how to configure security rules for your VPC. * **OCSP and port 80 requirements**: While Snowflake communications use port 443, Online Certificate Status Protocol (OCSP) certificate checks are transmitted over port 80. If port 80 is not open in your network, you may encounter OCSP-related issues, such as JDBC Error 5. To resolve this, ensure your network administrator opens the firewall to traffic on ports 443 and 80 and permits all URLs in the Snowflake allowlist. No customer data is transferred over unencrypted HTTP; port 80 is used strictly for OCSP operations. For more information, see [Common connectivity issues and resolutions](https://docs.snowflake.com/en/user-guide/client-connectivity-troubleshooting/common-issues) in the Snowflake documentation.
- Kafka cluster credentials. The following lists the different ways you can provide credentials. - Enter an existing [service account](../service-account.md#s3-cloud-service-account) resource ID. - Create a Confluent Cloud [service account](../service-account.md#s3-cloud-service-account) for the connector. Make sure to review the ACL entries required in the [service account documentation](../service-account.md#s3-cloud-service-account). Some connectors have specific ACL requirements. - Create a Confluent Cloud API key and secret. To create a key and secret, you can use [confluent api-key create](https://docs.confluent.io/confluent-cli/current/command-reference/api-key/confluent_api-key_create.html) *or* you can autogenerate the API key and secret directly in the Cloud Console when setting up the connector. ### Using the Confluent Cloud Console #### Step 1: Launch your Confluent Cloud cluster To create and launch a Kafka cluster in Confluent Cloud, see [Create a kafka cluster in Confluent Cloud](../../get-started/index.md#cloud-create-kafka-cluster). #### Step 2: Add a connector In the left navigation menu, click **Connectors**. If you already have connectors in your cluster, click **+ Add connector**. #### Step 3: Select your connector Click the **Snowflake Source** connector card. ![Snowflake Source Connector Card](images/ccloud-snowflake-source-icon.png) #### Step 4: Enter the connector details #### NOTE * Make sure you have all your [prerequisites](#cc-snowflake-source-prereqs) completed. * An asterisk ( \* ) designates a required entry. At the **Snowflake Source Connector** screen, complete the following: ### Kafka access 1. Select the way you want to provide **Kafka Cluster credentials**. You can choose one of the following options: - **My account**: This setting allows your connector to globally access everything that you have access to. With a user account, the connector uses an API key and secret to access the Kafka cluster. This option is not recommended for production. - **Service account**: This setting limits the access for your connector by using a [service account](../service-account.md#s3-cloud-service-account). This option is recommended for production. - **Use an existing API key**: This setting allows you to specify an API key and a secret pair. You can use an existing pair or create a new one. This method is not recommended for production environments. #### NOTE Freight clusters support only service accounts for Kafka authentication. 2. Click **Continue**. ### Authentication 1. Configure the authentication properties: **Authentication method** - **Credentials Source**: The source of the credentials for authentication. Use one of the following supported values: * `PRIVATE_KEY`: Authenticate using the private key for the Snowflake user. Enter only the part of the key between `--BEGIN RSA PRIVATE KEY--` and `--END RSA PRIVATE KEY--`. * `PRIVATE_KEY_PASSPHRASE`: Authenticate using both the private key and passphrase of the encrypted private key. - **Use secret manager**: Fetch sensitive configuration values from a secret manager. **Secret manager configuration** - **Secret manager**: Select the secret manager to use for retrieving sensitive data. - **Configurations from Secret manager**: Select the configurations whose values Confluent Cloud should fetch from the secret manager. - **Provider Integration**: Select an existing provider integration that has access to your secret manager. **Connect to your Snowflake instance** - **Snowflake Connection URL**: The Snowflake connection URL. Use the format `_.snowflakecomputing.com` or if your account is in the AWS US West (Oregon) region, use `.snowflakecomputing.com`. - **Snowflake User**: The user for the Snowflake instance. - **Snowflake Private Key**: The private key for the Snowflake user. - **Private Key Passphrase**: The passphrase of the encrypted private key. 2. Click **Continue**. ### Configuration **Output messages** - **Select output record value format**: Select the output Kafka record value format (data going to the Kafka topic). Valid values are AVRO, JSON_SR, or PROTOBUF. [Schema Registry](../../get-started/schema-registry.md#cloud-sr-config) must be enabled to use a Schema Registry-based format (for example, AVRO, JSON Schema, or PROTOBUF). **Name your topic(s)** - **Topic Prefix**: A logical name to prepend to table names to generate the Apache Kafka® topic name, where the connector publishes data. **Connector Details** - **Mode**: Set one of the following modes for updating a table each time it is polled: * **bulk**: Performs a bulk load of all eligible table each time it is polled. * **incrementing**: Uses a strictly incrementing column on each table to detect only new rows. Only rows with non-null value of incrementing column will be captured. * **Table to incrementing column mappings**: Enter a comma-separated list of table regex to incrementing column mappings. The expected format is `regex1:col2,regex2:col1`. Regexes will be matched against the fully-qualified table names. Identifier names are case-sensitive. Every table included for capture should match exactly one of the provided mappings. An example of valid input would be `COMPANY.EMPLOYEES.SALARY*:EMP_ID, COMPANY.FINANCE.ACCOUNTS.*:ID`. * **timestamp**: Uses timestamp column(s) to detect new and modified rows. On specifying multiple timestamp columns, the COALESCE SQL function will be used to find out the effective timestamp for a row. This assumes that the effective timestamp is updated with each write and its values are monotonically incrementing, but not necessarily unique. Only rows with non-null value of effective timestamp will be captured. * **Table to timestamp columns mappings**: Enter a comma-separated list of table regex to timestamp column mappings. Timestamp columns supplied should strictly be of the ``TIMESTAMP_NTZ`` type. When specifying multiple timestamp columns, the COALESCE SQL function will be used to find out the effective timestamp for a row. The expected format is `regex1:[col1|col2],regex2:[col3]`. Regexes will be matched against the fully-qualified table names. Identifier names are case-sensitive. Every table included for capture should match exactly one of the provided mappings. An example of valid input would be `COMPANY.EMPLOYEES.SALARY.*:[UPDATED_AT|MODIFIED_AT], COMPANY.FINANCE.ACCOUNTS.*:[CHANGED_AT]`. * **timestamp+incrementing**: Uses two columns, a timestamp column that detects new and modified rows, and a strictly incrementing column, which provides a globally unique ID for updates, so each row can be assigned a unique stream offset. Only rows with non-null effective timestamp value and non-null incrementing column value will be captured. - **Tables Included**: A comma-separated list of regular expressions to match the fully-qualified names of tables to copy. For example, `DB_A.PUBLIC.CUSTOMER.*,DB_B.PUBLIC.CUSTOMER.*,...``. - **Tables Excluded**: A comma-separated list of regular expressions to match the fully-qualified names of tables for the connector to ignore. For example, `DB_A.PUBLIC.CUSTOMER.*,DB_B.PUBLIC.CUSTOMER.*,...``. Note that this property applies only to tables filtered by the `table.include.list` field. - **Table to timestamp columns mappings**: A comma-separated list of table regex to timestamp columns mappings. Timestamp columns supplied should strictly be of type ``TIMESTAMP_NTZ``. On specifying multiple timestamp columns, COALESCE SQL function would be used to find out the effective timestamp for a row. Expected format is `regex1:[col1|col2],regex2:[col3]`. Regexes would be matched against the fully-qualified table names. Identifier names are case sensitive. Every table included for capture should match exactly one of the provided mappings. An example for a valid input would be `COMPANY.EMPLOYEES.SALARY.*:[UPDATED_AT|MODIFIED_AT], COMPANY.FINANCE.ACCOUNTS.*:[CHANGED_AT]`. - **Table to incrementing column mappings**: A comma-separated list of table regex to incrementing column mappings. Expected format is `regex1:col2,regex2:col1`. Regexes would be matched against the fully-qualified table names. Identifier names are case sensitive. Every table included for capture should match exactly one of the provided mappings. An example for a valid input would be `COMPANY.EMPLOYEES.SALARY*:EMP_ID,COMPANY.FINANCE.ACCOUNTS.*:ID`. - **Database Timezone**: The timezone to use to interpret the values for timestamp types that don’t include timezone information. This should be set to the timezone of the Snowflake account. **Data encryption** - Enable **Client-Side Field Level Encryption** for data encryption. Specify a **Service Account** to access the Schema Registry and associated encryption rules or keys with that schema. For more information on CSFLE or CSPE setup, see [Manage encryption for connectors](../csfle.md#connect-csfle). ### **Show advanced configurations** - **Schema context**: Select a schema context to use for this connector, if using a schema-based data format. This property defaults to the **Default** context, which configures the connector to use the default schema set up for Schema Registry in your Confluent Cloud environment. A schema context allows you to use separate schemas (like schema sub-registries) tied to topics in different Kafka clusters that share the same Schema Registry environment. For example, if you select a non-default context, a **Source** connector uses only that schema context to register a schema and a **Sink** connector uses only that schema context to read from. For more information about setting up a schema context, see [What are schema contexts and when should you use them?](../../sr/faqs-cc.md#faq-schema-contexts). **Additional Configs** - **Value Converter Decimal Format**: Specifies the `JSON` or `JSON_SR` serialization format for Connect `DECIMAL` logical type values with two allowed literals: `BASE64` to serialize `DECIMAL` logical types as base64 encoded binary data, and `NUMERIC` to serialize `DECIMAL` logical type values in `JSON` or `JSON_SR` as a number representing the decimal value. - **Key Converter Schema ID Serializer**: The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers. - **Value Converter Reference Subject Name Strategy**: Sets the subject reference name strategy for values. Valid entries are `DefaultReferenceSubjectNameStrategy` or `QualifiedReferenceSubjectNameStrategy`. You can use this strategy only with `PROTOBUF` format; the default strategy is `DefaultReferenceSubjectNameStrategy`. - **Errors Tolerance**: Use this property to configure the connector’s error handling behavior. #### WARNING Use this property with caution for sink connectors, as it can lead to data loss. If you set this property to `all`, the connector does not fail on errant records, but logs them (and sends to DLQ for sink connectors) and continues processing. If you set this property to `none`, the connector task fails on errant records. - **Value Converter Connect Meta Data**: Enables the Connect converter to add its metadata to the output schema. Applies to Avro converters. - **Value Converter Value Subject Name Strategy**: Determines how to construct the subject name under which the value schema is registered with Schema Registry. - **Key Converter Key Subject Name Strategy**: Determines how to construct the subject name for key schema registration. - **Value Converter Ignore Default For Nullables**: When set to `true`, this property ensures that the corresponding record in Kafka is `null`, instead of showing the default column value. Applies to the `AVRO`, `PROTOBUF`, and `JSON_SR` converters. - **Value Converter Schema ID Serializer**: The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers. **Auto-restart policy** - **Enable Connector Auto-restart**: Enables the auto-restart behavior of the connector and its task in the event of user-actionable errors. Defaults to `true`, enabling the connector to automatically restart in case of user-actionable errors. Set this property to `false` to disable auto-restart for failed connectors. If disabled, you must manually restart the connector. **Connector Details** - **Initial Timestamp**: The epoch timestamp in milliseconds which provides the start timestamp from where to capture rows. The value -1 sets the start timestamp to the current time. In this case, older data is not fetched. If not specified, the start timestamp is treated as epoch start time, and all data is fetched. Once the connector has successfully recorded a source offset, this property has no effect even if changed to a different value later. - **Table Types**: By default, the connector only detects tables with type `TABLE` from the source database. This configuration allows a comma-separated list of table types to extract. - **Timestamp granularity for timestamp columns**: Defines the granularity of the timestamp column. Defaults to `NANOS_LONG`. * `CONNECT_LOGICAL`: Represents timestamp values using Kafka Connect’s built-in representations. This approach may lead to precision loss, as it only supports millisecond precision. * `NANOS_LONG`: Represents timestamp values as nanoseconds since the epoch. * `NANOS_STRING`: Represents timestamp values as nanoseconds since the epoch, encoded as strings. - **Poll Interval (ms)**: The frequency in milliseconds (ms) for polling new data in each table. Defaults to 5000 ms (5 seconds). - **Quote SQL Identifiers**: When to quote table names, column names, and other identifiers in SQL statements. Use `always` when working with case-sensitive identifiers or reserved keywords. The default value is `never`. - **Numeric Mapping**: Map `NUMERIC` values to Connect types based on their precision, and optionally map them to specific integral or decimal types. - Use `none` if Connect’s `DECIMAL` logical type represents all `NUMERIC` columns. - Use `best_fit` if Confluent Cloud should cast `NUMERIC` columns to Connect’s `INT8`, `INT16`, `INT32`, `INT64`, or `FLOAT64` based on the column’s precision and scale. **Transforms** - **Single Message Transformations**: To add a new SMT, see [Add transforms](../single-message-transforms.md#cc-single-message-transforms-ui). For more information about unsupported SMTs, see [Unsupported transformations](../single-message-transforms.md#cc-single-message-transforms-unsupported-transforms). **Processing position** - **Set offsets**: Click **Set offsets** to define a specific offset for this connector to begin procession data from. For more information on managing offsets, see [Manage offsets](../offsets.md#connect-custom-offsets). For all property values and definitions, see [Configuration Properties](#cc-snowflake-source-config-properties). - Click **Continue**. ### Sizing Based on the number of topic partitions you select, you will be provided with a recommended number of tasks. 1. To change the number of recommended tasks, enter the number of tasks for the connector to use in the **Tasks** field. 2. Click **Continue**. ### Review and launch 1. Verify the connection details by previewing the running configuration. ```none { "config": { "kafka.auth.mode": "SERVICE_ACCOUNT", "kafka.service.account.id": "sa-dev123", "schema.context.name": "default", "value.subject.name.strategy": "TopicNameStrategy", "value.converter.reference.subject.name.strategy": "DefaultReferenceSubjectNameStrategy", "connector.class": "SnowflakeSource", "name": "SnowflakeSourceConnector_0", "connection.url": "cflt.snowflakecomputing.com", "connection.user": "", "connection.credentials.source": "PRIVATE_KEY", "connection.private.key": "", "topic.prefix": "cli.", "mode": "timestamp", "timestamp.columns.mapping": "TEST_DB.PUBLIC.TEST_TS_MODE.*:[UPDATED_AT]", "db.timezone": "UTC", "table.types": "TABLE", "timestamp.granularity": "NANOS_LONG", "poll.interval.ms": "5000", "output.data.format": "AVRO", "tasks.max": "1", "auto.restart.on.user.error": "true" } } ``` 2. After you’ve validated that the properties are configured to your satisfaction, click **Launch**. The status for the connector should go from **Provisioning** to **Running**. ![Check the connector status](images/ccloud-snowflake-source-status.png) #### Step 5: Check the Kafka topic After the connector is running, verify that messages are populating your Kafka topic. For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](../connect-api-section.md#ccloud-connect-api) section. ### Using the Confluent CLI Complete the following steps to set up and run the connector using the Confluent CLI. #### NOTE Make sure you have all your [prerequisites](#cc-snowflake-source-prereqs) completed. #### Step 1: List the available connectors Enter the following command to list available connectors: ```none confluent connect plugin list ``` #### Step 2: List the connector configuration properties Enter the following command to show the connector configuration properties: ```none confluent connect plugin describe ``` The command output shows the required and optional configuration properties. #### Step 3: Create the connector configuration file Create a JSON file that contains the connector configuration properties. The following example shows the required connector properties. ```none { "name" : "SnowflakeSource_cli", "connector.class": "SnowflakeSource", "kafka.auth.mode": "KAFKA_API_KEY", "kafka.api.key": "", "table.include.list": "TEST_DB.PUBLIC.TEST_TS_MODE.*", "table.exclude.list": "TEST_DB.PUBLIC.TEST_TS_MODE_.*", "db.timezone": "America/Los_Angeles", "mode": "timestamp", "timestamp.columns.mapping": "TEST_DB.PUBLIC.TEST_TS_MODE.*:[UPDATED_AT]", "topic.prefix": "cli.", "connection.url": "locator.snowflakecomputing.com", "connection.user": "", "connection.credentials.source": "PRIVATE_KEY", "connection.private.key": "", "tasks.max": "1", "output.data.format": "AVRO" } ``` Note the following property definitions: * `"name"`: Sets a name for your new connector. * `"connector.class"`: Identifies the connector plugin name. * `"kafka.auth.mode"`: Identifies the connector authentication mode you want to use. There are two options: `SERVICE_ACCOUNT` or `KAFKA_API_KEY` (the default). To use an API key and secret, specify the configuration properties `kafka.api.key` and `kafka.api.secret`, as shown in the example configuration (above). To use a [service account](../service-account.md#s3-cloud-service-account), specify the **Resource ID** in the property `kafka.service.account.id=`. To list the available service account resource IDs, use the following command: ```bash confluent iam service-account list ``` For example: ```bash confluent iam service-account list Id | Resource ID | Name | Description +---------+-------------+-------------------+------------------- 123456 | sa-l1r23m | sa-1 | Service account 1 789101 | sa-l4d56p | sa-2 | Service account 2 ``` * `"topic.prefix"`: Enter a topic prefix. The connector automatically creates Kafka topics using the naming convention: ``. The tables are created with the properties: `topic.creation.default.partitions=1` and `topic.creation.default.replication.factor=3`. If you want to create topics with specific settings, create the topics before running this connector. If you are configuring granular access using a service account, you must set up [ACLs for the topic prefix](../service-account.md#cloud-service-account-jdbc-mongo-acls). * `"output.data.format"`: Sets the output Kafka record value format (data coming from the connector). Valid entries are **AVRO**, **JSON_SR**, or **PROTOBUF**. You must have Confluent Cloud Schema Registry configured if using a schema-based message formats. * `"db.timezone"`: Specify the timezone to interpret the values for timestamp types that don’t include timezone information. This should be set to the timezone of the Snowflake account.. For more information, see this [list of database timezones](https://en.wikipedia.org/wiki/List_of_tz_database_time_zones). #### NOTE To enable CSFLE or CSPE for data encryption, specify the following properties: * `csfle.enabled`: Flag to indicate whether the connector honors CSFLE or CSPE rules. * `sr.service.account.id`: A Service Account to access the Schema Registry and associated encryption rules or keys with that schema. For more information on CSFLE or CSPE setup, see [Manage encryption for connectors](../csfle.md#connect-csfle). **SMTs**: For details about adding SMTs using the Confluent CLI, see the [Single Message Transformations](../single-message-transforms.md#cc-single-message-transforms) documentation. See [Configuration Properties](#cc-snowflake-source-config-properties) for all properties and definitions. #### Step 4: Load the properties file and create the connector Enter the following command to load the configuration and start the connector: ```none confluent connect cluster create --config-file .json ``` For example: ```none confluent connect cluster create --config-file snowflake-source.json ``` Example output: ```none Created connector SnowflakeSource_0 lcc-ix4dl ``` #### Step 5: Check the connector status Enter the following command to check the connector status: ```none confluent connect cluster list ``` Example output: ```none ID | Name | Status | Type +-----------+-------------------------+---------+-------+ lcc-ix4dl | SnowflakeSource_0 | RUNNING | source ``` #### Step 6: Check the Kafka topic. After the connector is running, verify that messages are populating your Kafka topic. For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](../connect-api-section.md#ccloud-connect-api) section. ## Configuration Properties Use the following configuration properties with the fully managed connector. For self-managed connector property definitions and other details, see the connector docs in [Self-managed connectors for Confluent Platform](/platform/current/connect/kafka_connectors.html). ### Kafka Cluster credentials `kafka.auth.mode` : Kafka Authentication mode. It can be one of KAFKA_API_KEY or SERVICE_ACCOUNT. It defaults to KAFKA_API_KEY mode, whenever possible.
* Type: string * Valid Values: SERVICE_ACCOUNT, KAFKA_API_KEY * Importance: high `kafka.api.key` : Kafka API Key. Required when kafka.auth.mode==KAFKA_API_KEY.
* Type: password * Importance: high `kafka.service.account.id` : The Service Account that will be used to generate the API keys to communicate with Kafka Cluster.
* Type: string * Importance: high `kafka.api.secret` : Secret associated with Kafka API key. Required when kafka.auth.mode==KAFKA_API_KEY.
* Type: password * Importance: high ### Schema Config `schema.context.name` : Add a schema context name. A schema context represents an independent scope in Schema Registry. It is a separate sub-schema tied to topics in different Kafka clusters that share the same Schema Registry instance. If not used, the connector uses the default schema configured for Schema Registry in your Confluent Cloud environment.
* Type: string * Default: default * Importance: medium ### How should we connect to your data? `name` : Sets a name for your connector.
* Type: string * Valid Values: A string at most 64 characters long * Importance: high ### Authentication method `connection.credentials.source` : The source of the credentials to use for authentication. Supported values are: PRIVATE_KEY: Use the private key to authenticate. PRIVATE_KEY_PASSPHRASE: Use the private key and passphrase to authenticate.
* Type: string * Default: PRIVATE_KEY * Importance: high `secret.manager.enabled` : Fetch sensitive configuration values from a secret manager.
* Type: boolean * Default: false * Importance: high ### Secret manager configuration `secret.manager` : Select the secret manager to use for retrieving sensitive data.
* Type: string * Importance: high `secret.manager.managed.configs` : Select the configurations to fetch their values from the secret manager.
* Type: list * Importance: high `secret.manager.provider.integration.id` : Select an existing provider integration that has access to your secret manager.
* Type: string * Importance: high ### How should we connect to your Snowflake Instance? `connection.url` : Snowflake connection URL. Supported formats are _.snowflakecomputing.com or If the account is located in the AWS US West (Oregon) region, then .snowflakecomputing.com.
* Type: string * Importance: high `connection.user` : User to be used for authenticating to snowflake.
* Type: string * Importance: high `connection.private.key` : Private key for Snowflake user.
* Type: password * Importance: high `connection.private.key.passphrase` : Passphrase of the encrypted private key.
* Type: password * Importance: high ### How should we name your topic(s)? `topic.prefix` : Prefix to prepend to table names to generate the name of the Kafka topic to publish data to.
* Type: string * Importance: high ### Connector Details `mode` : Mode represents the criteria on which table is polled each time. Options include – BULK: perform a bulk load of all the eligible tables each time it is polled. TIMESTAMP: use timestamp column(s) to detect new and modified rows. On specifying multiple timestamp columns, COALESCE SQL function would be used to find out the effective timestamp for a row. This assumes that the effective timestamp is updated with each write, it’s values are monotonically incrementing, but not necessarily unique. Only rows with non null value of effective timestamp would be captured. INCREMENTING: use a strictly incrementing column on each table to detect only new rows. Only rows with non null value of incrementing column would be captured. TIMESTAMP AND INCREMENTING: use two columns, a timestamp column that detects new and modified rows and a strictly incrementing column which provides a globally unique ID for updates so each row can be assigned a unique stream offset. Only rows with non null effective timestamp value and non null incrementing column value would be captured
* Type: string * Importance: medium `table.include.list` : A comma-separated list of regular expressions that match the fully-qualified names of tables to be copied. Identifier names are case sensitive. For example, `table.include.list: "DB_A.PUBLIC.CUSTOMER.*,DB_B.PUBLIC.CUSTOMER.*,"`.
* Type: list * Importance: medium `table.exclude.list` : A comma-separated list of regular expressions that match the fully-qualified names of tables not to be copied. This only applies on the tables filtered using include list. Identifier names are case sensitive. For example, `table.exclude.list: "DB_A.PUBLIC.CUSTOMER.*,DB_B.PUBLIC.CUSTOMER.*,"`.
* Type: list * Importance: medium `timestamp.columns.mapping` : A comma-separated list of table regex to timestamp columns mappings. Timestamp columns supplied should strictly be of type `TIMESTAMP_NTZ`. On specifying multiple timestamp columns, COALESCE SQL function would be used to find out the effective timestamp for a row. Expected format is `regex1:[col1|col2],regex2:[col3]`. Regexes would be matched against the fully-qualified table names. Identifier names are case sensitive. Every table included for capture should match exactly one of the provided mappings. An example for a valid input would be `COMPANY.EMPLOYEES.SALARY.*:[UPDATED_AT|MODIFIED_AT], COMPANY.FINANCE.ACCOUNTS.*:[CHANGED_AT]`.
* Type: list * Importance: medium `incrementing.column.mapping` : A comma-separated list of table regex to incrementing column mappings. Expected format is `regex1:col2,regex2:col1`. Regexes would be matched against the fully-qualified table names. Identifier names are case sensitive. Every table included for capture should match exactly one of the provided mappings. An example for a valid input would be `COMPANY.EMPLOYEES.SALARY*:EMP_ID,COMPANY.FINANCE.ACCOUNTS.*:ID`.
* Type: list * Importance: medium `db.timezone` : Timezone to be used when interpreting values for timestamp types which don’t have a timezone information in them. This should be set to the timezone of the snowflake account.
* Type: string * Importance: medium `timestamp.initial` : Epoch timestamp in milliseconds which provides the start timestamp from where to capture rows. The value -1 sets the start timestamp to the current time, in this case older data would not be fetched. If not specified, start timestamp is treated as epoch start time, hence all data is fetched. Once the connector has managed to successfully record a source offset, this property has no effect even if changed to a different value later on.
* Type: long * Importance: medium `table.types` : By default, the connector will only detect tables with type TABLE. This config allows a command separated list of table types to extract.
* Type: list * Default: TABLE * Importance: medium `timestamp.granularity` : Define the granularity of the Timestamp column. CONNECT_LOGICAL: Represents timestamp values using Kafka Connect built-in representations. This may lead to loss of precision as this only supports milliseconds precision. NANOS_LONG: Represents timestamp values as nanos since epoch. Avoid this if any of the eligible timestamp columns contains timestamps greater than 2262-11-04 23:47:16.854775807 GMT, nanos since epoch value would not fit in the range for long and hence would lead to erroneous values. Use nanos_string in such cases. NANOS_STRING: represents timestamp values as nanos since epoch in string.
* Type: string * Default: NANOS_LONG * Importance: low `poll.interval.ms` : Frequency in ms to poll for new data in each table.
* Type: int * Default: 5000 (5 seconds) * Valid Values: [100,…] * Importance: low `quote.sql.identifiers` : When to quote table names, column names, and other identifiers in SQL statements. Use always when working with case-sensitive identifiers or reserved keywords. Default value is never.
* Type: string * Default: never * Importance: low `numeric.mapping` : Map NUMERIC values by precision and optionally scale to integral or decimal types. Use none if all NUMERIC columns are to be represented by Connect’s DECIMAL logical type. Use best_fit if NUMERIC columns should be cast to Connect’s INT8, INT16, INT32, INT64, or FLOAT64 based upon the column’s precision and scale.
* Type: string * Default: none * Importance: low ### Output messages `output.data.format` : Sets the output Kafka record value format. Valid entries are AVRO, JSON_SR, PROTOBUF. Note that you need to have Confluent Cloud Schema Registry configured if using a schema-based message format like AVRO, JSON_SR, and PROTOBUF.
* Type: string * Default: AVRO * Importance: high ### Number of tasks for this connector `tasks.max` : Maximum number of tasks for the connector.
* Type: int * Valid Values: [1,…] * Importance: high ### Additional Configs `header.converter` : The converter class for the headers. This is used to serialize and deserialize the headers of the messages.
* Type: string * Importance: low `producer.override.compression.type` : The compression type for all data generated by the producer. Valid values are none, gzip, snappy, lz4, and zstd.
* Type: string * Importance: low `producer.override.linger.ms` : The producer groups together any records that arrive in between request transmissions into a single batched request. More details can be found in the documentation: [https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#linger-ms](https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#linger-ms).
* Type: long * Valid Values: [100,…,1000] * Importance: low `value.converter.allow.optional.map.keys` : Allow optional string map key when converting from Connect Schema to Avro Schema. Applicable for Avro Converters.
* Type: boolean * Importance: low `value.converter.auto.register.schemas` : Specify if the Serializer should attempt to register the Schema.
* Type: boolean * Importance: low `value.converter.connect.meta.data` : Allow the Connect converter to add its metadata to the output schema. Applicable for Avro Converters.
* Type: boolean * Importance: low `value.converter.enhanced.avro.schema.support` : Enable enhanced schema support to preserve package information and Enums. Applicable for Avro Converters.
* Type: boolean * Importance: low `value.converter.enhanced.protobuf.schema.support` : Enable enhanced schema support to preserve package information. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.flatten.unions` : Whether to flatten unions (oneofs). Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.generate.index.for.unions` : Whether to generate an index suffix for unions. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.generate.struct.for.nulls` : Whether to generate a struct variable for null values. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.int.for.enums` : Whether to represent enums as integers. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.latest.compatibility.strict` : Verify latest subject version is backward compatible when use.latest.version is true.
* Type: boolean * Importance: low `value.converter.object.additional.properties` : Whether to allow additional properties for object schemas. Applicable for JSON_SR Converters.
* Type: boolean * Importance: low `value.converter.optional.for.nullables` : Whether nullable fields should be specified with an optional label. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.optional.for.proto2` : Whether proto2 optionals are supported. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.scrub.invalid.names` : Whether to scrub invalid names by replacing invalid characters with valid characters. Applicable for Avro and Protobuf Converters.
* Type: boolean * Importance: low `value.converter.use.latest.version` : Use latest version of schema in subject for serialization when auto.register.schemas is false.
* Type: boolean * Importance: low `value.converter.use.optional.for.nonrequired` : Whether to set non-required properties to be optional. Applicable for JSON_SR Converters.
* Type: boolean * Importance: low `value.converter.wrapper.for.nullables` : Whether nullable fields should use primitive wrapper messages. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.wrapper.for.raw.primitives` : Whether a wrapper message should be interpreted as a raw primitive at root level. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `errors.tolerance` : Use this property if you would like to configure the connector’s error handling behavior. WARNING: This property should be used with CAUTION for SOURCE CONNECTORS as it may lead to dataloss. If you set this property to ‘all’, the connector will not fail on errant records, but will instead log them (and send to DLQ for Sink Connectors) and continue processing. If you set this property to ‘none’, the connector task will fail on errant records.
* Type: string * Default: none * Importance: low `key.converter.key.schema.id.serializer` : The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers.
* Type: string * Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer * Importance: low `key.converter.key.subject.name.strategy` : How to construct the subject name for key schema registration.
* Type: string * Default: TopicNameStrategy * Importance: low `value.converter.decimal.format` : Specify the JSON/JSON_SR serialization format for Connect DECIMAL logical type values with two allowed literals:
BASE64 to serialize DECIMAL logical types as base64 encoded binary data and
NUMERIC to serialize Connect DECIMAL logical type values in JSON/JSON_SR as a number representing the decimal value.
* Type: string * Default: BASE64 * Importance: low `value.converter.flatten.singleton.unions` : Whether to flatten singleton unions. Applicable for Avro and JSON_SR Converters.
* Type: boolean * Default: false * Importance: low `value.converter.ignore.default.for.nullables` : When set to true, this property ensures that the corresponding record in Kafka is NULL, instead of showing the default column value. Applicable for AVRO,PROTOBUF and JSON_SR Converters.
* Type: boolean * Default: false * Importance: low `value.converter.reference.subject.name.strategy` : Set the subject reference name strategy for value. Valid entries are DefaultReferenceSubjectNameStrategy or QualifiedReferenceSubjectNameStrategy. Note that the subject reference name strategy can be selected only for PROTOBUF format with the default strategy being DefaultReferenceSubjectNameStrategy.
* Type: string * Default: DefaultReferenceSubjectNameStrategy * Importance: low `value.converter.value.schema.id.serializer` : The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers.
* Type: string * Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer * Importance: low `value.converter.value.subject.name.strategy` : Determines how to construct the subject name under which the value schema is registered with Schema Registry.
* Type: string * Default: TopicNameStrategy * Importance: low ### Auto-restart policy `auto.restart.on.user.error` : Enable connector to automatically restart on user-actionable errors.
* Type: boolean * Default: true * Importance: medium ## Supported Data Types The connector creates change events for table changes. Each change event mirrors the table’s schema, with a field for every column value. The data type of each table column determines how the connector represents the column values in the corresponding change event fields. For certain data types, such as `TIMESTAMP_NTZ` data type, you can customize how the connector maps them by modifying the default configuration settings. This allows more control over handling various data types, ensuring that the change events reflect the desired format and meet specific requirements. #### WARNING When using timestamp mode, the Snowflake Source connector only supports `TIMESTAMP_NTZ` columns for the columns specified in `timestamp.column.mapping` property. The connector assumes data in these columns always uses the server’s time zone, which is defined by the `db.timezone` property. For the columns specified in the `timestamp.column.mapping property`, if you manually insert data using a different time zone, such as UTC, than the server’s time zone, the connector might incorrectly interpret the timestamps and show incorrect behavior. You should ensure that values in the columns specified in `timestamp.column.mapping` are auto-populated, and set the `db.timezone` property to correctly match the server’s timezone value. ### Numeric data types The following table describes how the connector maps numeric types. | Snowflake data type | Connect type | Notes | |-------------------------------------------------------|----------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------| | NUMBER, DECIMAL, DEC, NUMERIC | BYTES | `org.apache.kafka.connect.data.Decimal`

The `scale` schema parameter contains an integer that represents how many digits the decimal point shifted. | | INT, INTEGER, BIGINT, SMALLINT, TINYINT, BYTEINT | BYTES | `org.apache.kafka.connect.data.Decimal`

The `scale` schema parameter contains an integer that represents how many digits the decimal point shifted. | | FLOAT, FLOAT4, FLOAT8, DOUBLE, DOUBLE PRECISION, REAL | FLOAT64 | | #### NOTE You should specify in advance the precision and scale you expect for the values when creating a column. For example, NUMBER(3) limits it to only three-digit integers. This practice can help enforce a certain degree of integrity on the values entered into that column. ### String and binary data types The following table describes how the connector maps string and binary types. | Snowflake data type | Connect type | |-------------------------------------------------------------------------|----------------| | VARCHAR, STRING, TEXT, NVARCHAR, NVARCHAR2, CHAR VARYING, NCHAR VARYING | STRING | | CHAR, CHARACTER, NCHAR | STRING | | BINARY | BYTES | ### Logical data types The following table describes how the connector maps logical types. | Snowflake data type | Connect type | |-----------------------|----------------| | BOOLEAN | BOOLEAN | ### Date and time data types The following table describes how the connector maps date and time types. | Snowflake data type | Connect type | Notes | |-------------------------------------|-----------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | DATE | INT32 | `org.apache.kafka.connect.data.Date`

Represents the number of days since the epoch. | | TIME (P) | INT64 | `io.confluent.connect.snowflake.data.NanoTime`

Represents the number of nanoseconds past midnight, and does not include timezone information. | | TIMESTAMP_NTZ (P)/ DATETIME (P) | INT64 OR STRING | Depending on configuration property timestamp.granularity, appropriate Kafka Connect type is chosen.

For `NANOS_STRING`, the connector uses `STRING`.

For `NANOS_LONG`, the connector uses `INT_64`.

For `CONNECT_LOGICAL`, the connector uses `org.apache.kafka.connect.data.Timestamp`. | | TIMESTAMP_LTZ (P), TIMESTAMP_TZ (P) | STRING | `io.confluent.connect.snowflake.data.ZonedTimestamp`
A string representation of a timestamp with timezone information. | ### Semi-structured data types The following table describes how the connector maps semi-structured types. | Snowflake data type | Connect type | |-----------------------|----------------| | VARIANT | STRING | | OBJECT | STRING | | ARRAY | STRING | ## Frequently asked questions Find answers to frequently asked questions about the Snowflake Source connector for Confluent Cloud. ### Why is my connector producing duplicate data in an infinite loop? When using `timestamp` mode, the connector may repeatedly extract the same records, causing duplicate data and infinite polling. The following are common causes: * Timezone mismatch: The most common cause is a mismatch between the `db.timezone` configuration and your Snowflake account’s actual timezone. If `db.timezone` does not match your Snowflake account’s actual timezone, the connector may misinterpret timestamp values and repeatedly fetch the same records. * Manual timestamp insertion: If you manually insert timestamp values using a timezone different from your server’s timezone, the connector may show incorrect behavior. ### Why does my connector fail with `JWT_TOKEN_INVALID_PUBLIC_KEY_FINGERPRINT_MISMATCH` errors? This error occurs when the public key fingerprint in the JWT token does not match the one stored in Snowflake. Error code 394304: ```none JWT_TOKEN_INVALID_PUBLIC_KEY_FINGERPRINT_MISMATCH ``` The following are common causes and solutions: * Private key mismatch: The private key configured in the connector does not match the public key assigned to the Snowflake user. Verify you are using the correct key pair. * Key pair not assigned to user: The public key may not be properly assigned to your Snowflake user account. Run `DESC USER ` in Snowflake to verify the `RSA_PUBLIC_KEY_FP` property matches your key. * Key regeneration: If you recently regenerated your key pair, ensure you updated both the connector configuration and the Snowflake user’s public key. To resolve this issue, verify your key pair matches by checking the public key fingerprint in Snowflake and regenerating the keys if necessary. See [Generate a Snowflake key pair](#cc-snowflake-db-source-gen-key-pair). ### Why are Snowflake `NUMBER` columns appearing as `BYTES` in Kafka? Snowflake `NUMBER` data types are mapped to `BYTES` in Kafka Connect using the `org.apache.kafka.connect.data.Decimal` schema type. This is expected behavior — the `NUMBER` type can represent arbitrary precision decimal values, and Kafka Connect uses `BYTES` to preserve that precision. To control how `NUMBER` columns are mapped, set the `numeric.mapping` configuration property: * `none` (default): Maps `NUMBER` to `BYTES` using the `org.apache.kafka.connect.data.Decimal` schema type. * `best_fit`: Attempts to map `NUMBER` columns to a native numeric type (`INT8`, `INT16`, `INT32`, `INT64`, or `FLOAT64`) based on the column’s precision and scale. Use this option when your `NUMBER` columns contain integer or floating-point values that fit within these types. For full property details, see [Configuration Properties](#cc-snowflake-source-config-properties). For more information about data type mapping, see [Supported Data Types](#cc-snowflake-source-data-types). ### How do I configure timestamp mode correctly? Timestamp mode uses timestamp columns to detect new and modified rows. Configure these properties carefully to avoid data duplication. The following configuration is required: * `db.timezone`: Set this to match your Snowflake account’s timezone exactly. To find your Snowflake account timezone, run `SHOW PARAMETERS LIKE 'TIMEZONE'` in Snowflake. * `timestamp.column.mapping`: Specify which columns to use for timestamp tracking. For the correct format and examples, see [Configuration Properties](#cc-snowflake-source-config-properties). * `TIMESTAMP_NTZ` columns only: The connector only supports `TIMESTAMP_NTZ` columns for `timestamp` mode. These columns must be auto-populated and use the server’s timezone. Example configuration: ```none "mode": "timestamp", "db.timezone": "America/Los_Angeles", "timestamp.column.mapping": "TEST_DB.PUBLIC.MY_TABLE.*:[UPDATED_AT]" ``` Consider the following: * Do not manually insert timestamp values using a different timezone than your server’s timezone. * Ensure the timestamp column values are monotonically increasing. * For multiple timestamp columns, use the `COALESCE` SQL function format: `"timestamp.column.mapping": "DB.SCHEMA.TABLE.*:[COL1,COL2]"`. For more information, see the warning in [Supported Data Types](#cc-snowflake-source-data-types). ### What should I set for the `poll.interval.ms` property? The `poll.interval.ms` property controls how frequently the connector queries Snowflake for changes. Recommended values depend on your use case: * Low-latency requirements: Set to 1000-5000 ms for near real-time data capture. Default is 5000 ms. * High-volume tables: Use longer intervals such as 30000-60000 ms to reduce load on Snowflake and avoid excessive queries. * Batch processing: Set to higher values such as 300000 ms or more if you process data in batches. Consider the following: * Shorter intervals increase load on your Snowflake account and may incur higher costs. * Very short intervals may not improve latency if your tables are not updated frequently. * The connector only fetches changes since the last poll, so you will not get duplicate data with longer intervals. Monitor your connector’s throughput and Snowflake query costs to determine the optimal polling interval for your workload. ### Why am I seeing duplicate messages after a connector restart? Duplicate messages can occur when the connector enters a degraded state or after task reconfiguration. When connector tasks are rebalanced or reconfigured, the connector may reprocess some records from the last successful offset. This is expected behavior for at-least-once delivery semantics. ### How do I handle Snowflake `VARIANT`, `OBJECT`, and `ARRAY` types? The connector maps Snowflake semi-structured data types to `STRING` in Kafka Connect. The following Snowflake types are converted to STRING: ```none VARIANT → STRING OBJECT → STRING ARRAY → STRING ``` To work with these types, consider the following: * Parse in consumer: The data is serialized as a JSON string. Parse the string in your consumer application to extract the structured data. * Cast in Snowflake: If you need structured data in Kafka, create a view that casts or extracts specific fields from `VARIANT`/ `OBJECT`/ `ARRAY` columns into primitive types. * Schema evolution: Changes to the structure of `VARIANT`, `OBJECT`, and `ARRAY` data do not trigger schema evolution in the connector. Example Snowflake view: ```none CREATE VIEW my_table_flat AS SELECT id, variant_col:field1::STRING AS field1, variant_col:field2::NUMBER AS field2 FROM my_table; ``` Use the view as the source for your connector instead of the base table. For more information, see [Supported Data Types](#cc-snowflake-source-data-types). ## Next Steps For an example that shows fully managed Confluent Cloud connectors in action with Confluent Cloud for Apache Flink, see the [Cloud ETL Demo](/platform/current/tutorials/examples/cloud-etl/docs/index.html). This example also shows how to use Confluent CLI to manage your resources in Confluent Cloud. [![image](images/topology.png)](https://docs.confluent.io/platform/current/tutorials/examples/cloud-etl/docs/index.html)