<a id="cc-snowflake-sink"></a>

# Snowflake Sink Connector for Confluent Cloud

The fully managed Snowflake Sink connector for Confluent Cloud writes records from
Apache Kafka® topics directly into Snowflake tables, exposing the data to services
for querying, enrichment, and analytics. The connector supports Avro, JSON
Schema, Protobuf, JSON (schemaless), or BYTES (schemaless) input data.

The Confluent Cloud connector plugin name is `SnowflakeSink`. The underlying
connector class is `com.snowflake.kafka.connector.SnowflakeSinkConnector`.

#### NOTE
* This Quick Start is for the fully managed Confluent Cloud connector. If you are
  installing the connector locally for Confluent Platform, see the [Snowflake Connector for
  Kafka](https://docs.snowflake.com/en/user-guide/kafka-connector.html)
  documentation.
* 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 Sink connector (version 3.4.0) provides the following features:

* **Database authentication:** The connector supports two authentication methods:
  - **Private key authentication** (default): Uses RSA key pair authentication for secure access to Snowflake.
    This method supports both **SNOWPIPE** and **SNOWPIPE_STREAMING** ingestion methods.
  - **OAuth 2.0 (Bring your own app)**: Uses Snowflake OAuth with custom client credentials for authentication.
    This method only supports the **SNOWPIPE_STREAMING** ingestion method. Because OAuth uses the authorization code grant
    flow, this authentication method is only available via the Confluent Cloud Console and is not
    supported via the Confluent CLI or Confluent REST API. For more information, see
    [Snowflake OAuth for Custom Clients](https://docs.snowflake.com/en/user-guide/oauth-custom).
* **Proxy support:** The connector supports connecting to Snowflake through an HTTPS proxy.
* **OCSP certificate validation control:** The connector allows you to disable Online Certificate Status Protocol (OCSP)
  certificate revocation checks using the `snowflake.disable.ocsp.checks` configuration property. Enable this property only in environments where OCSP servers are not accessible. By default, OCSP checks are enabled.
* **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 the [connector configuration](#cc-snowflake-sink-setup-connection).
* **Snowflake Ingestion Methods:**  The connector supports the [Snowpipe](https://docs.snowflake.com/en/user-guide/data-load-snowpipe-intro)
  (default) and [Snowpipe Streaming for Kafka](https://docs.snowflake.com/en/user-guide/data-load-snowpipe-streaming-kafka)
  data ingestion methods. Using Snowpipe Streaming may provide a [cost-benefit](https://www.snowflake.com/blog/data-ingestion-best-practices-part-three/)
  for your Snowflake project.
* **Snowflake Schematization** (`snowflake.enable.schematization`): When set to `TRUE` the connector
  provides schema detection and evolution when using [Snowpipe Streaming for
  Kafka](https://docs.snowflake.com/en/user-guide/data-load-snowpipe-streaming-kafka).
  The default value is `FALSE`. For more information, see [Schema detection
  and schema evolution](https://docs.snowflake.com/en/user-guide/data-load-snowpipe-streaming-kafka#schema-detection-and-schema-evolution).
* **Input data formats:** The connector supports Avro, JSON Schema, Protobuf, JSON (schemaless), or BYTES (schemaless) input data formats. [Schema Registry](../../get-started/schema-registry.md#cloud-sr-config) must be enabled to use a Schema Registry-based format (for example, Avro, JSON_SR (JSON Schema), or Protobuf).
* **Select configuration properties:** The following properties determine what metadata is included in the `RECORD_METADATA` column in the Snowflake database table.
  - `snowflake.metadata.createtime`: If this value is set to `false`, the `CreateTime` property value is omitted from the metadata in the `RECORD_METADATA` column. The default value is `true`.
  - `snowflake.metadata.topic`: If this value is set to `false`, the `topic` property value is omitted from the metadata in the `RECORD_METADATA` column. The default value is `true`.
  - `snowflake.metadata.offset.and.partition`: If the value is set to false, the `Offset` and `Partition` property values are omitted from the metadata in the `RECORD_METADATA` column. The default value is `true`.
  - `snowflake.streaming.metadata.connectorPushTime`: If this value is set to `true`, the `SnowflakeConnectorPushTime` property value is included in the metadata in the `RECORD_METADATA` column. The default value is `false`. This metadata is only applicable when using the SNOWPIPE_STREAMING ingestion method.
  - `snowflake.metadata.all`: If the value is set to false, the metadata in the `RECORD_METADATA` column is empty. The default value is `true`.
* **Load data into Iceberg tables:**  The connector now supports ingesting data into Snowflake-managed
  [Apache Iceberg table](https://docs.snowflake.com/en/user-guide/tables-iceberg).
  To enable this feature, set `snowflake.streaming.iceberg.enabled` to `true`.
  Note that Iceberg table support is available only when using the SNOWPIPE_STREAMING ingestion method.
* **Secret manager integration**: The connector supports secret manager integration. For `Password` based authentication, the connector can retrieve the following configurations from an integrated secret manager at runtime as needed.

  | **Secret manager managed configuration**   | **Type**   |
  |--------------------------------------------|------------|
  | `snowflake.user.name`                      | `STRING`   |
  | `snowflake.private.key`                    | `PASSWORD` |
  | `snowflake.role.name`                      | `STRING`   |
  | `snowflake.url.name`                       | `STRING`   |
  | `snowflake.private.key.passphrase`         | `PASSWORD` |

  For more information, see [Create a secret manager integration in Confluent Cloud](../secret-manager-integration/overview.md#cloud-secret-manager-quickstart).

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 Sink Connector](../limits.md#cc-snowflake-sink-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).

<a id="cc-snowflake-table-suffixes"></a>

## Target table naming guidelines

Note the following table naming guidelines and limitations:

* The fully managed Snowflake Sink connector allows you to configure `topic:table` name mapping. This feature is also supported by the [self-managed Snowflake Sink connector](https://docs.snowflake.com/en/user-guide/kafka-connector-overview.html#target-tables-for-kafka-topics).
* Snowflake itself has limitations on object (table) naming conventions. See [Identifier Requirements](https://docs.snowflake.com/en/sql-reference/identifiers-syntax.html#identifier-requirements) for details.
* Kafka is much more permissive with topic naming conventions. You are allowed to use Kafka topic names that break the table name mapping in the Confluent Cloud Snowflake Sink connector.

  When a Kafka topic name does not conform to Snowflake’s table
  naming limitations (for example, `my-topic-name`), the connector will rename
  the topic to a safe name with an appended hash (for example,
  `my_topic_name_021342`). A conforming topic name (for example,
  `my_topic_name`) will send results to the expected table named
  `my_topic_name`.
* If the connector needs to adjust the name of the table created for a Kafka topic, there is the potential for identical table names. For example, if you are reading data from Kafka topics `numbers+x` and `numbers-x`, the tables created for these topics will both be named `NUMBERS_X`. To avoid table name duplication, the connector appends a suffix to the table name. The suffix is an underscore followed by a generated hash.

<a id="cc-snowflake-db-sink-gen-key-pair"></a>

## 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='<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.

<a id="cc-snowflake-db-sink-oauth-setup"></a>

## Set up OAuth 2.0 authentication

To use OAuth 2.0 authentication with the Snowflake Sink connector, you need to
create a security integration in Snowflake. This section provides the steps to
configure OAuth 2.0.

#### IMPORTANT
**OAuth 2.0 Limitations:**

* **Ingestion method**: OAuth 2.0 authentication only supports the **SNOWPIPE_STREAMING**
  ingestion method. If you need to use the SNOWPIPE ingestion method, you
  must use private key authentication.
* **Configuration access**: Because OAuth uses the authorization code grant flow, you can only configure OAuth
  authentication via the Cloud Console.
  This authentication method is not available through the Confluent CLI or
  Confluent REST API.

### Prerequisites for OAuth 2.0

* You must have `ACCOUNTADMIN` role or equivalent privileges in Snowflake.
* A Snowflake role that the connector will use (for example, `KAFKA_CONNECTOR_ROLE`).

### Create OAuth security integration

1. Connect to Snowflake using a client with `ACCOUNTADMIN` role.
2. Create a security integration for OAuth. Run the following SQL command in
   your Snowflake worksheet:
   ```sql
   USE ROLE ACCOUNTADMIN;
   CREATE OR REPLACE SECURITY INTEGRATION confluent_kafka_oauth
     TYPE = OAUTH
     ENABLED = TRUE
     OAUTH_CLIENT = CUSTOM
     OAUTH_CLIENT_TYPE = 'CONFIDENTIAL'
     OAUTH_REDIRECT_URI = 'https://confluent.cloud/api/connect/oauth/callback'
     OAUTH_ISSUE_REFRESH_TOKENS = TRUE
     OAUTH_REFRESH_TOKEN_VALIDITY = 7776000;
   ```

   #### NOTE
   * `confluent_kafka_oauth` is a placeholder name. You can use any name
     that follows Snowflake naming conventions.
   * `OAUTH_REFRESH_TOKEN_VALIDITY` is set to 7776000 seconds (90 days),
     which is the maximum allowed validity period. You can set this to a
     lower value based on your security requirements. The minimum is 3600
     seconds (1 hour).
3. Grant usage on the security integration to the role the connector will use:
   ```sql
   GRANT USAGE ON INTEGRATION confluent_kafka_oauth TO ROLE <your_connector_role>;
   ```

   Replace `<your_connector_role>` with the actual role name you want to use
   with the connector.
4. Retrieve the OAuth client ID and client secret by running:
   ```sql
   SELECT SYSTEM$SHOW_OAUTH_CLIENT_SECRETS('confluent_kafka_oauth');
   ```

   This command returns a JSON object containing the `OAUTH_CLIENT_ID` and
   `OAUTH_CLIENT_SECRET`. Save these values as you will need them when
   configuring the connector.

   Example output:
   ```json
   {
     "OAUTH_CLIENT_ID": "abcd1234EFGH5678==",
     "OAUTH_CLIENT_SECRET": "xyz9876aBcDeFgHiJkL=="
   }
   ```

For more information about Snowflake OAuth, see the [Snowflake OAuth FAQ](https://community.snowflake.com/s/article/FAQs-Snowflake-OAuth).

## Quick Start

Use this quick start to get up and running with the Confluent Cloud Snowflake Sink
connector. The quick start provides the basics of selecting the connector and
configuring it to consume data from Kafka and persist the data to a Snowflake
database.

<a id="cc-snowflake-db-sink-prereqs"></a>

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).
  - [Schema Registry](../../get-started/schema-registry.md#cloud-sr-config) must be enabled to use a Schema Registry-based format (for example, Avro, JSON_SR (JSON Schema), or Protobuf).
  - One of the following authentication methods configured in Snowflake:
    - **Private key authentication** (default): A Snowflake [account and key pair](#cc-snowflake-db-sink-gen-key-pair) to use for connector authentication with the Snowflake database.
      This method supports both **SNOWPIPE** and **SNOWPIPE_STREAMING** ingestion methods.
    - **OAuth 2.0**: [OAuth security integration](#cc-snowflake-db-sink-oauth-setup) configured in Snowflake with the OAuth client ID and client secret.
      This method only supports the **SNOWPIPE_STREAMING** ingestion method and and must be configured through the Cloud Console. Note that it is not supported via the Confluent CLI or Confluent REST API.
  - The [user created](#cc-snowflake-db-sink-gen-key-pair) must be granted privileges in Snowflake to modify the database and schema. For more information, see [Access Control Privileges](https://docs.snowflake.com/en/user-guide/security-access-control-privileges.html).
  - For networking considerations, see [Networking and DNS](../overview.md#connect-internet-access-resources). 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).
  - **OCSP and port 80 requirements**: While Snowflake communications use port 443, 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.
  <br/>
    If you cannot open port 80 or your environment prevents access to OCSP servers, set `snowflake.disable.ocsp.checks` to `true`. Because this bypasses certificate revocation validation, enable this setting only if your network security policy allows it and no other connectivity options are available. 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.
  <br/>
  - 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 Sink** connector card.

![Snowflake Sink Connector Card](images/ccloud-snowflake-sink-icon.png)

<a id="cc-snowflake-sink-setup-connection"></a>

#### Step 4: Enter the connector details

#### NOTE
* Ensure you have all your [prerequisites](#cc-snowflake-db-sink-prereqs) completed.
* The example commands use Confluent CLI version 2. For more information
  see, [Confluent CLI v2](https://docs.confluent.io/confluent-cli/current/migrate.html#cli-migrate).

At the **Add Snowflake Sink Connector** screen, complete the following:

### Topic selection

If you’ve already populated your Kafka topics, select the topics you want
to connect from the **Topics** list.

To create a new topic, click **+Add new topic**.

### 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**
   - **Authentication method**: Under **Snowflake Credentials**, select your **Authentication method**:
     - **Password** (default): Use private key authentication
     - **OAuth 2.0 (Bring your own app)**: Use Snowflake OAuth authentication
   - **Use secret manager**: Enable this setting to fetch sensitive configuration values, such as the `Password`, from a secret manager.

   **Secret manager configuration**
   - **Secret manager**: Select the secret manager that Confluent Cloud should use to retrieve sensitive data.
   - **Configurations from Secret manager**: Select the configurations whose values Confluent Cloud should fetch from the secret manager.
   - **Provider Integration**: Select an existing integration that has access to your resource such as the secret
     manager.

   **Snowflake Credentials**
   - **Connection URL**: The URL for accessing your Snowflake
     account. Use the format `https://<account-locator>.<region-ID>.snowflakecomputing.com`. For example:

     `https://abcd12345.us-east-1.snowflakecomputing.com`

     The `https://` and `443` port number are optional. For more
     information, see [Account Locator in a Region](https://docs.snowflake.com/en/user-guide/admin-account-identifier.html#option-2-account-locator-in-a-region).
     Do not use the region ID if your account is in the AWS US West
     region and you are using [AWS PrivateLink](https://docs.snowflake.com/en/user-guide/admin-security-privatelink.html).
   - **Connection user name**: The [username created earlier](#cc-snowflake-db-sink-gen-key-pair).
   - **Snowflake role**: Specifies the role the connector uses when inserting rows into the table.
     - For OAuth 2.0 authentication, this property is always required.
     - For Private Key authentication, this property is required if the active ingestion method is `SNOWPIPE_STREAMING`. If the ingestion method is `SNOWPIPE`, the default role assigned to the user is used.
   - **Private key**: The [private key created earlier](#cc-snowflake-db-sink-gen-key-pair) as a single line. Enter only the
     part of the key between `--BEGIN RSA PRIVATE KEY--` and `--END RSA
     PRIVATE KEY--`.
   - **OAuth Client ID**: Enter the client ID obtained from the
     [OAuth security integration](#cc-snowflake-db-sink-oauth-setup)
     you created in Snowflake.
   - **Client Secret**: Enter the client secret obtained from the OAuth
     security integration.
   - **Decryption key of private key**: Enter the decryption key for the private key to
     decrypt it. Use this field only if the private key is encrypted.
   - **Enable HTTPS proxy for Snowflake connections**: Set to `true` to
     use an HTTPS proxy to connect to Snowflake. The default is `false`.
     When set to `true`, the following properties are available:
     - *HTTPS proxy host for Snowflake connections*: The hostname or IP address of the proxy server.
     - *HTTPS proxy port for Snowflake connections*: The port number of the proxy server.
     - *Hosts to bypass HTTPS proxy for Snowflake connections*: A list of hosts that should be reached directly, bypassing the proxy.
     - *HTTPS proxy username for Snowflake connections*: The username for authenticating with the proxy server.
     - *HTTPS proxy password for Snowflake connections*: The password for authenticating with the proxy server.
   - **HTTPS proxy host for Snowflake connections**: The HTTPS proxy host for Snowflake connections.
   - **HTTPS proxy port for Snowflake connections**: The HTTPS proxy port for Snowflake connections.
   - **Disable OCSP checks**: If set to `true`, disables Online Certificate Status Protocol (OCSP) certificate revocation checks for Snowflake JDBC connections. Enable this only in environments where OCSP servers are not accessible. The default value is `false`.
   - **Hosts to bypass HTTPS proxy for Snowflake connections**: The hosts to bypass HTTPS proxy for Snowflake connections.
   - **HTTPS proxy username for Snowflake connections**: The HTTPS proxy username for Snowflake connections.
   - **HTTPS proxy password for Snowflake connections**: The HTTPS proxy password for Snowflake connections.

   **Snowflake Database Details**
   - **Database name**: The database name containing the table to
     insert rows into.
   - **Schema name**: The schema name that contains the table to insert rows into.
   - **Topics to tables mapping**: A comma-separated list of tuples specifying the mapping of
     topics to specific tables. For example, to map topic-1 to table-1 and
     topic-2 to table-2, use the format: `<topic-1>:<table-1>`, `<topic-2>:<table-2>`, and so on for additional mappings.
2. Click **Continue**.

### Configuration

#### NOTE
Configuration properties that are not shown in the
Cloud Console use the default values. See
[Configuration Properties](#cc-snowflake-sink-config-properties) for configuration property
values and descriptions.

- **Input Kafka record value format**: Select the input Kafka record value format (data coming from the
  Kafka topic). Valid values are JSON (schemaless), AVRO, JSON_SR (JSON Schema), or PROTOBUF.
  A valid schema must be available in [Schema Registry](../../get-started/schema-registry.md#cloud-sr-config) to use a schema-based message format (for example,
  Avro, JSON_SR (JSON Schema), or Protobuf).

**Snowflake connection**

- **Ingestion method**: Select the ingestion method. This property defaults to
  `SNOWPIPE`. Select `SNOWPIPE_STREAMING`. [Snowpipe Streaming for
  Kafka](https://docs.snowflake.com/en/user-guide/data-load-snowpipe-streaming-kafka)
  may provide a [cost-benefit](https://www.snowflake.com/blog/data-ingestion-best-practices-part-three/)
  for your Snowflake project.

**Data decryption**

- Enable **Client-Side Field Level Encryption** for
  data decryption. Specify a **Service Account** to
  access the Schema Registry and associated encryption rules or keys with that schema. Select the connector behavior
  (`ERROR` or `NONE`) on data decryption failure. If set to `ERROR`, the connector fails and writes the encrypted data
  in the DLQ. If set to `NONE`, the connector writes the encrypted data in the target system without decryption.
  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).
- **Input Kafka record key format**: Select the Input Kafka record key format. Valid values are JSON
  (schemaless), AVRO, JSON_SR (JSON Schema), PROTOBUF, or STRING.
  A valid schema must be available in [Schema Registry](../../get-started/schema-registry.md#cloud-sr-config)
  to use a schema-based message format (for example, Avro, JSON_SR
  (JSON Schema), or Protobuf).
  for additional information.
- **Key Converter Reference Subject Name Strategy**: Set the subject reference name strategy for key. Valid values are `DefaultReferenceSubjectNameStrategy` or `QualifiedReferenceSubjectNameStrategy`. Note that the subject reference name strategy can be selected only for PROTOBUF format with the default strategy being `DefaultReferenceSubjectNameStrategy`.

**Additional Configs**

- **Value Converter Replace Null With Default**: Specifies whether to replace fields that have a default value and that are null to the default value. When set to `true`, the connector uses the default value; otherwise, it uses `null`. Applies to the `JSON` converter.
- **Value Converter Schema ID Deserializer**: Sets the class name of the schema ID deserializer for values. The deserializer reads schema IDs from 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`.
- **Schema ID For Value Converter**: Sets the schema ID to use for deserialization when using `ConfigSchemaIdDeserializer`. This lets you specify a fixed schema ID for deserializing message values. This property is applicable only when `value.converter.value.schema.id.deserializer` is set to `ConfigSchemaIdDeserializer`.
- **Value Converter Schemas Enable**: Includes schema within each of the serialized values. Input messages must contain `schema` and `payload` fields and must not contain additional fields. For plain `JSON` data, set this to `false`. Applies to the `JSON` converter.
- **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.
- **Key Converter Schema ID Deserializer**: Sets the class name of the schema ID deserializer for keys. The deserializer reads schema IDs from message headers.
- **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.
- **Schema GUID For Key Converter**: Sets the schema GUID to use for deserialization when using `ConfigSchemaIdDeserializer`. This lets you specify a fixed schema GUID for deserializing message keys. This property is applicable only when `key.converter.key.schema.id.deserializer` is set to `ConfigSchemaIdDeserializer`.
- **Schema GUID For Value Converter**: Sets the schema GUID to use for deserialization when using `ConfigSchemaIdDeserializer`. This lets you specify a fixed schema GUID for deserializing message values. This property is applicable only when `value.converter.value.schema.id.deserializer` is set to `ConfigSchemaIdDeserializer`.
- **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.
- **Schema ID For Key Converter**: Sets the schema ID to use for deserialization when using `ConfigSchemaIdDeserializer`. This lets you specify a fixed schema ID for deserializing message keys. This property is applicable only when `key.converter.key.schema.id.deserializer` is set to `ConfigSchemaIdDeserializer`.

**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.

**Consumer configuration**

- **Max poll interval(ms)**: Sets the maximum delay between subsequent consume requests to Kafka. Use this property to
  improve connector performance in cases when the connector cannot send records to the sink system.
  The default is 300,000 milliseconds (5 minutes).
- **Max poll records**: Sets the maximum number of records to consume from Kafka in a single request. Use this property to
  improve connector performance in cases when the connector cannot send records to the sink system.
  The default is 500 records.

**Connection details**

- **Whether or not to include “createtime” in metadata**: If the
  value is set to `FALSE`, the `CreateTime` property value is omitted
  from the metadata in the `RECORD_METADATA` column. The default value
  is `TRUE`.
- **Whether or not to include “topic” in metadata**: If the value is
  set to `FALSE`, the topic property value is omitted from the
  metadata in the `RECORD_METADATA` column. The default value is
  `TRUE`.
- **Whether or not to include “offset” and “partition” in metadata**: If the value is set to `FALSE`, the Offset and Partition
  property values are omitted from the metadata in the
  `RECORD_METADATA` column. The default value is `TRUE`.
- **Whether or not to include “connectorPushTime” in metadata**: If the value is set to `TRUE`, the `ConnectorPushTime` property value is added to the metadata in the `RECORD_METADATA` column. The default value is `FALSE`. This works with only `SNOWPIPE_STREAMING` ingestion mode.
- **Whether or not to include metadata column**: If the value is set to `FALSE`, the metadata in the `RECORD_METADATA` column is completely empty. The default value is `TRUE`.
- **Enable Schematization**: Specify to `TRUE` to enable schema detection and evolution for Kafka connector with Snowpipe Streaming. The default value is `FALSE`.
- **The time in seconds to flush cached data**: Number of seconds between buffer flushes, where the flush is from the Kafka’s memory cache to the internal stage. The default value is 120 seconds. Minimum value allowed is 10 for snowflake.ingestion.method=SNOWPIPE and 1 for snowflake.ingestion.method=SNOWPIPE_STREAMING. The connector uses buffer.count.records and buffer.size.bytes=10,000,000 (10MB) as well. Whichever comes first, the connector will flush Kafka records to Snowflake.
- **The number of records to flush cached data**: The number of
  records between buffer flushes, where the flush is from the
  Kafka’s memory cache to the internal stage. The default and
  minimum value is 10,000 records. The connector also uses the
  previous time in seconds property and the followingsize of
  the buffer property to determine when to flush cached data. When
  one of these property values is reached, the connector flushes
  Kafka records to Snowflake.
- **The size of the connector record buffer**: The buffer size
  defaults to 5000000 bytes (5 MB). The records are compressed when
  written to Snowflake.

  When a flush is triggered when the cache reaches 5 MB, you might
  expect to see a 5 MB data file in Snowflake. You will see a much
  smaller file (for example, ~150 KB). This is because the 5 MB of
  flushed data is converted from Java to UTF. This conversion
  reduces the file size by 50 percent. The file is then compressed
  with gzip, which further reduces the file size by 95 percent.
- **Enable Iceberg Ingestion**: Interested in integrating with Iceberg? Take a look at Tableflow — a solution designed to simplify and streamline your Iceberg workflows, refer [https://www.confluent.io/product/tableflow/](https://www.confluent.io/product/tableflow/).
  This property specifies whether the connector ingests data into an Iceberg table. The connector fails if this property doesn’t match the actual table type, refer [https://docs.snowflake.com/en/user-guide/kafka-connector-iceberg](https://docs.snowflake.com/en/user-guide/kafka-connector-iceberg).
- **Enable Task Failure on Authorization Errors**: If set to true, the connector task will fail whenever it encounters an authorization error from Snowflake.
- **Enable dynamic flush**: If set to `true`, each connector task can consume up to twice as many topic
  partitions based on the buffer size. This option is useful for workloads with
  many topic partitions but low throughput per partition.

  #### NOTE
  Enabling this configuration can degrade performance. Because each task
  handles more partitions, flush operations can trigger more often with
  smaller batches of data based on the connector’s buffer footprint.
- **Max client lag in seconds for Snowpipe Streaming**: Determines how often the connector flushes the buffer in the Ingest SDK.
  The default value is `30` seconds. Setting this value too low can degrade
  performance, while setting it too high can increase latency. This
  configuration applies only when `snowflake.ingestion.method` is set to
  `SNOWPIPE_STREAMING`.
- **Enable Streaming channel names prefixed with connector name**: When enabled, the Snowflake Streaming channel names are prefixed with the
  connector name. This configuration enables or disables the use of channel
  names that were used solely in Kafka Connector versions `2.1.0` and
  `2.1.1`, and is intended for users who previously used these versions and
  have not updated the connector since.

  #### IMPORTANT
  Enabling this configuration in any other case results in data
  duplication.

  This configuration applies only when `snowflake.ingestion.method` is set
  to `SNOWPIPE_STREAMING`.
- **Enable handling infinity values**: Whether to enable correct handling of infinity and negative infinity values
  in float and double fields when using Snowpipe Streaming. When enabled,
  infinity values are correctly ingested as `Inf` or `-Inf`. When disabled,
  infinity values are treated as `NaN` for backward compatibility. This
  configuration applies only when `snowflake.ingestion.method` is set to
  `SNOWPIPE_STREAMING`.

**Error handling**

- **Error Tolerance**: Connector error tolerance. Supported values are none, all

**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).

- Click **Continue**.

### Sizing

Based on the number of topic partitions you select, you will be provided
with a recommended number of tasks. One task can handle up to 100
partitions.

1. To change the number of recommended tasks, enter the number of
   [tasks](/platform/current/connect/concepts.html#tasks) for the connector to use
   in the **Tasks** field. Each task is limited to a number of topic
   partitions based on the `buffer.size.bytes` property value. For
   example, a `10` MB buffer size is limited to 50 topic partitions, a
   `20` MB buffer is limited to 25 topic partitions, `50` MB buffer is
   limited to 10 topic partitions, and a `100` MB buffer to 5 topic
   partitions.
2. Click **Continue**.

### Review and launch

Review the configuration summary and verify the following:

1. Verify the connection details and click **Launch**.
   ![Launch the connector](images/ccloud-snowflake-sink-launch-connector.png)

   The status for the connector should go from **Provisioning** to **Running**.
   ![Check the connector status](images/ccloud-snowflake-sink-status.png)
2. Click **Launch**.

#### Step 5: Check Snowflake

After the connector is running, verify that messages are populating your Snowflake database table.

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.

For Snowflake troubleshooting, see [Troubleshooting Issues](https://docs.snowflake.com/en/user-guide/kafka-connector-ts.html#troubleshooting-issues) in the Snowflake documentation.

#### NOTE
* The Snowflake Sink connector does not remove Snowflake pipes when a connector is deleted. For instructions to manually clean up Snowflake pipes, see [Dropping Pipes](https://docs.snowflake.com/en/user-guide/kafka-connector-manage.html#dropping-pipes).
* [Snowflake Snowpipe](https://docs.snowflake.com/en/user-guide/data-load-snowpipe-intro.html) failure can prevent messages from showing up in the target table despite being successfully written by the Snowflake Sink connector. If this happens, check the Snowflake [COPY_HISTORY](https://docs.snowflake.com/en/sql-reference/functions/copy_history.html) view, internal stage, or table stage to find the message and associated error. For more on the workflow of Snowflake Sink connector, see [Workflow for the Kafka Connector](https://docs.snowflake.com/en/user-guide/kafka-connector-overview.html#workflow-for-the-kafka-connector).

### 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-db-sink-prereqs) completed.
* OAuth 2.0 authentication is **not supported** via the Confluent CLI. If you need to
  use OAuth authentication, you must create the connector using the
  Confluent Cloud Console. The Confluent CLI only supports private key
  authentication.

#### 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 <connector-plugin-name>
```

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 examples show the required connector properties for each authentication
method.

**Example 1: Private Key Authentication (default)**

```json
{
  "connector.class": "SnowflakeSink",
  "name": "<connector-name>",
  "kafka.auth.mode": "KAFKA_API_KEY",
  "kafka.api.key": "<my-kafka-api-key>",
  "kafka.api.secret": "<my-kafka-api-secret>",
  "topics": "<topic1>, <topic2>",
  "input.data.format": "JSON",
  "snowflake.url.name": "https://wm83168.us-central1.gcp.snowflakecomputing.com:443",
  "snowflake.user.name": "<login-username>",
  "snowflake.private.key": "<private-key>",
  "snowflake.database.name": "<database-name>",
  "snowflake.schema.name": "<schema-name>",
  "tasks.max": "1"
}
```

**Example 2: OAuth 2.0 Authentication (Reference Only)**

OAuth 2.0 authentication is not supported via the Confluent CLI or Confluent REST API. Because OAuth
uses the authorization code grant flow, you must configure OAuth authentication through the Confluent Cloud Console.
The example below shows the configuration properties set when using OAuth via the Cloud Console
and is provided for reference only.

```json
{
  "connector.class": "SnowflakeSink",
  "name": "<connector-name>",
  "kafka.auth.mode": "KAFKA_API_KEY",
  "kafka.api.key": "<my-kafka-api-key>",
  "kafka.api.secret": "<my-kafka-api-secret>",
  "topics": "<topic1>, <topic2>",
  "input.data.format": "JSON",
  "authentication.method": "OAuth 2.0 (Bring your own app)",
  "snowflake.url.name": "https://wm83168.us-central1.gcp.snowflakecomputing.com:443",
  "snowflake.user.name": "<login-username>",
  "snowflake.role.name": "<your-connector-role>",
  "snowflake.ingestion.method": "SNOWPIPE_STREAMING",
  "oauth.client.id": "<oauth-client-id>",
  "oauth.client.secret": "<oauth-client-secret>",
  "oauth.refresh.token": "<oauth-refresh-token>",
  "snowflake.database.name": "<database-name>",
  "snowflake.schema.name": "<schema-name>",
  "tasks.max": "1"
}
```

#### NOTE
If you need to use OAuth 2.0 authentication, you must create the connector
using the Confluent Cloud Console. Use **Connect with
Snowflake** option that handles the OAuth authorization code grant flow and
automatically populates the refresh token.

Note the following required property definitions:

* `"connector.class"`: Identifies the connector plugin name.
* `"name"`: Enter a name for your connector.

* `"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=<service-account-resource-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
  ```

* `"topics"`: Enter one topic or multiple comma-separated topics.
* `"input.data.format"`:  Sets the input Kafka record value format (data coming from the Kafka topic). Valid entries are **AVRO**, **JSON_SR**, **PROTOBUF**, **JSON**, or **BYTES**. You must have Confluent Cloud Schema Registry configured if using a schema-based message format (for example, Avro, JSON_SR (JSON Schema), or Protobuf).
* `"snowflake.url.name"`: The URL the connector uses to reach your Snowflake account, in the format `https://<account_locator>.<region_id>.<cloud_provider>.snowflakecomputing.com:443`. The `https://` prefix and `443` port number are optional. For more information, see [Account Locator in a Region](https://docs.snowflake.com/en/user-guide/admin-account-identifier.html#option-2-account-locator-in-a-region). Do not use the region ID if your account is in the AWS US West region and you are not using [AWS PrivateLink](https://docs.snowflake.com/en/user-guide/admin-security-privatelink.html).
* `"snowflake.user.name"`: Enter the [user name created earlier](#cc-snowflake-db-sink-gen-key-pair). Note that if using the
  **SNOWPIPE_STEAMING** ingestion method, you must add the
  `"snowflake.role.name"` property. See
  [Configuration Properties](#cc-snowflake-sink-config-properties) for all property values and
  descriptions.
* `"snowflake.private.key"`:
  - Enter the [private key created earlier](#cc-snowflake-db-sink-gen-key-pair) as a **single line**.
  - Enter only the part of the key between `--BEGIN RSA PRIVATE KEY--` and `--END RSA PRIVATE KEY--`.
* `"snowflake.database.name"`: Enter the database name containing the table to insert rows into.
* `"snowflake.schema.name"`: Enter the Snowflake **Schema name** that contains the table to insert rows into.
* `"tasks.max"`: Enter the number of [tasks](/platform/current/connect/concepts.html#tasks) for the connector. Refer to Confluent Cloud [connector limitations](../limits.md#cc-snowflake-sink-limits) for additional information.

**OAuth 2.0 Authentication Properties**

#### IMPORTANT
**OAuth 2.0 Limitations:**

* **Ingestion method**: OAuth 2.0 authentication only supports the **SNOWPIPE_STREAMING**
  ingestion method. You must set `"snowflake.ingestion.method":
  "SNOWPIPE_STREAMING"` when using OAuth authentication.
* **Configuration access**: Because OAuth uses the authorization code grant flow, you can only configure OAuth
  authentication via the Confluent Cloud Console.
  This authentication method is not available through the Confluent CLI or
  Confluent REST API.

The following properties are set when using OAuth 2.0 authentication
(`"authentication.method": "OAuth 2.0 (Bring your own app)"`). These
properties are automatically configured when you use the **Connect with
Snowflake** option in the Cloud Console:

* `"authentication.method"`: Set to `"OAuth 2.0 (Bring your own app)"` to
  use OAuth authentication. If not specified, the connector uses private key
  authentication (default).
* `"snowflake.ingestion.method"`: Must be set to `"SNOWPIPE_STREAMING"` when
  using OAuth 2.0 authentication. OAuth is not supported with the `"SNOWPIPE"`
  ingestion method.
* `"snowflake.role.name"`: The Snowflake role to use for the connection. This
  is **required** when using OAuth 2.0 authentication.
* `"oauth.client.id"`: The OAuth client ID from your [Snowflake security
  integration](#cc-snowflake-db-sink-oauth-setup). This is **required** when
  using OAuth 2.0.
* `"oauth.client.secret"`: The OAuth client secret from your Snowflake
  security integration. This is **required** when using OAuth 2.0.
* `"oauth.refresh.token"`: The OAuth refresh token obtained during the
  authentication flow. This token is used to automatically refresh access
  tokens. This is **required** when using OAuth 2.0.

  #### IMPORTANT
  The refresh token has a Time-To-Live (TTL) based on the
  `OAUTH_REFRESH_TOKEN_VALIDITY` setting in your Snowflake security
  integration. The minimum TTL is 1 hour (3600 seconds) and the maximum is
  90 days (7776000 seconds). When the refresh token expires, the connector
  will fail. For recovery steps, see [OAuth Refresh Token Expiration](#cc-snowflake-oauth-token-expiration).

<a id="cc-snowflake-sink-proxy-support"></a>

To configure the connector to use an HTTPS proxy, set the following optional properties:

* `"snowflake.useHttpsProxy"`: Set to `true` to use an HTTPS proxy. Defaults to `false`.
* `"snowflake.https.proxyHost"`: The hostname or IP address of the proxy server.
* `"snowflake.https.proxyPort"`: The port number of the proxy server.
* `"snowflake.https.proxyUser"`: The username for authenticating with the proxy server.
* `"snowflake.https.proxyPassword"`: The password for authenticating with the proxy server.
* `"snowflake.https.nonProxyHosts"`: A comma-separated list of hosts that should be reached directly, bypassing the proxy.

To disable OCSP certificate revocation checks in environments where OCSP servers are not accessible:

* `"snowflake.disable.ocsp.checks"`: Set to `true` to disable OCSP checks. The default is `false`. Enable this property only in secure environments where OCSP servers are unreachable.

The following are optional properties to include in the configuration. These
properties affect what metadata is included in the `RECORD_METADATA` column in
the Snowflake database table.

* `"snowflake.metadata.createtime"`: If this value is set to `"false"`, the `CreateTime` property value is omitted from the metadata in the `RECORD_METADATA` column. The default value is `"true"`.
* `"snowflake.metadata.topic"`: If this value is set to `"false"`, the `topic` property value is omitted from the metadata in the `RECORD_METADATA` column. The default value is `"true"`.
* `"snowflake.metadata.offset.and.partition"`: If the value is set to `"false"`, the `Offset` and `Partition` property values are omitted from the metadata in the `RECORD_METADATA` column. The default value is `"true"`.
* `"snowflake.streaming.metadata.connectorPushTime"`: If this value is set to `"true"`, the `SnowflakeConnectorPushTime` property value is included in the metadata in the `RECORD_METADATA` column. The default value is `"false"`. This metadata is only applicable when using the SNOWPIPE_STREAMING ingestion method.
* `"snowflake.metadata.all"`: If the value is set to `"false"`, the metadata in the `RECORD_METADATA` column is empty. The default value is `"true"`.

**Buffering and flushing configuration**

**Snowpipe mode**

Set the following properties that determine when records are flushed to Snowflake. Records are flushed when the first one of these values is met. For example: The interval to flush records is set to `120` seconds. This time interval has elapsed from the last flush, but the number of records value has not been met. Records are flushed because the time interval tripped before the records property.

* `"buffer.flush.time"`: The time (in seconds) the connector waits before flushing cached records to Snowflake. The default value is `120` seconds, and the minimum value is 10 seconds. You can configure a longer time interval.
* `"buffer.count.records"`: Records are cached in a buffer (per partition) before they are flushed to Snowflake. The default value is `10000`. This is the minimum number of records. You can configure this to a larger number of records. Records are flushed to Snowflake when the number of records reaches the property value.
* `"buffer.size.bytes"`: Records are cached in a buffer (per partition) before being written to
  Snowflake as data files. The buffer size defaults to `10000000` bytes (10 MB). This is the minimum
  cache size value. Records are flushed to Snowflake when this buffer reaches the property size.

  #### NOTE
  When a flush is triggered when the cache reaches 5 MB, you might expect to
  see a 5 MB data file in Snowflake. You will see a much smaller file (for
  example, ~150 KB). This is because the 5 MB of flushed data is converted
  from Java to UTF. This conversion reduces the file size by 50 percent. The
  file is then compressed with gzip, which further reduces the file size by
  95 percent.

**Snowpipe streaming mode**

In `SNOWPIPE_STREAMING` mode, the Snowflake Sink connector uses the Snowflake Ingest SDK to send records to Snowflake. The connector batches records and sends them based on two parameters:

* `"snowflake.streaming.max.client.lag"`: The maximum time (in seconds) the SDK waits before flushing buffered records to Snowflake.
  The default value is `30` seconds. The minimum value is `1` second and the maximum value is `300` seconds. This is the primary control for flush frequency in streaming mode.
  Note that reducing this value may increase the frequency of API calls to Snowflake, which can have implications for both performance and cost.
  For more information, see [Latency recommendations by Snowflake](https://docs.snowflake.com/en/user-guide/snowpipe-streaming/snowpipe-streaming-classic-recommendation#latency-recommendations).
* `"buffer.size.bytes"`: The maximum size in bytes of the channel buffer before data is flushed to Snowflake. The default value is `10000000` bytes (10 MB). This maps to the `MAX_CHANNEL_SIZE_IN_BYTES` parameter in the Snowflake Ingest SDK. When the channel buffer reaches this size, the SDK flushes data to Snowflake.

#### NOTE
The `buffer.flush.time` and `buffer.count.records` properties do **not** apply to `SNOWPIPE_STREAMING` mode.

**Setting tasks.max**

* `"tasks.max"`: Enter the maximum number of [tasks](/platform/current/connect/concepts.html#tasks) that the connector will use. Each task is limited to a number of topic partitions based on the `buffer.size.bytes` property value. For example, a `10` MB buffer size is limited to 50 topic partitions, a `20` MB buffer is limited to 25 topic partitions, `50` MB buffer is limited to 10 topic partitions, and a `100` MB buffer to 5 topic partitions.

#### 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.
* `csfle.onFailure`: Configures the connector behavior (`ERROR` or `NONE`) on data decryption failure.
  If set to `ERROR`, the connector fails and writes the encrypted data
  in the DLQ. If set to `NONE`, the connector writes the encrypted data in the target system without decryption.

When using CSFLE or CSPE with connectors that route failed messages to a Dead Letter Queue (DLQ),
be aware that data sent to the DLQ is written in unencrypted plaintext. This poses
a significant security risk as sensitive data that should be encrypted may be exposed in the DLQ.

Do not use DLQ with CSFLE or CSPE in the current version. If you need error handling for
CSFLE- or CSPE-enabled data, use alternative approaches such as:

* Setting the connector behavior to `ERROR` to throw exceptions instead of routing to DLQ
* Implementing custom error handling in your applications
* Using `NONE` to pass encrypted data through without decryption

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. For a list of SMTs that are not supported with this connector, see [Unsupported transformations](../single-message-transforms.md#cc-single-message-transforms-unsupported-transforms).

See [Configuration Properties](#cc-snowflake-sink-config-properties) for all property values and
descriptions.

#### 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 <file-name>.json
```

For example:

```none
confluent connect cluster create --config-file snowflake-sink.json
```

Example output:

```none
Created connector confluent-snowflake 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   | confluent-snowflake     | RUNNING | sink
```

#### Step 6: Check Snowflake

After the connector is running, verify that records are populating your Snowflake database.

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.

For Snowflake troubleshooting, see [Troubleshooting Issues](https://docs.snowflake.com/en/user-guide/kafka-connector-ts.html#troubleshooting-issues) in the Snowflake documentation.

#### NOTE
* The Snowflake Sink connector does not remove Snowflake pipes when a connector is deleted. For instructions to manually clean up Snowflake pipes, see [Dropping Pipes](https://docs.snowflake.com/en/user-guide/kafka-connector-manage.html#dropping-pipes).
* [Snowflake Snowpipe](https://docs.snowflake.com/en/user-guide/data-load-snowpipe-intro.html) failure can prevent messages from showing up in the target table despite being successfully written by the Snowflake Sink connector. If this happens, check the Snowflake [COPY_HISTORY](https://docs.snowflake.com/en/sql-reference/functions/copy_history.html) view, internal stage, or table stage to find the message and associated error. For more on the workflow of Snowflake Sink connector, see [Workflow for the Kafka Connector](https://docs.snowflake.com/en/user-guide/kafka-connector-overview.html#workflow-for-the-kafka-connector).

<a id="cc-snowflake-sink-config-properties"></a>

## 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).

### Which topics do you want to get data from?

`topics.regex`
: A regular expression that matches the names of the topics to consume from. This is useful when you want to consume from multiple topics that match a certain pattern without having to list them all individually.
  <br/>
  * Type: string
  * Importance: low

`topics`
: Identifies the topic name or a comma-separated list of topic names.
  <br/>
  * Type: list
  * Importance: high

`errors.deadletterqueue.topic.name`
: The name of the topic to be used as the dead letter queue (DLQ) for messages that result in an error when processed by this sink connector, or its transformations or converters. Defaults to ‘dlq-${connector}’ if not set. The DLQ topic will be created automatically if it does not exist. You can provide `${connector}` in the value to use it as a placeholder for the logical cluster ID.
  <br/>
  * Type: string
  * Default: dlq-${connector}
  * Importance: low

### 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.
  <br/>
  * Type: string
  * Default: default
  * Importance: medium

### Input messages

`input.data.format`
: Sets the input Kafka record value format. Valid entries are JSON, AVRO, JSON_SR, PROTOBUF, STRING or BYTES. Note that you need to have Confluent Cloud Schema Registry configured if using a schema-based message format like AVRO, JSON_SR, and PROTOBUF
  <br/>
  * Type: string
  * Default: JSON
  * Importance: high

`input.key.format`
: Sets the input Kafka record key format. Valid entries are AVRO, JSON_SR, PROTOBUF, STRING or JSON. Note that you need to have Confluent Cloud Schema Registry configured if using a schema-based message format like AVRO, JSON_SR, and PROTOBUF
  <br/>
  * Type: string
  * Default: STRING
  * Valid Values: AVRO, JSON, JSON_SR, PROTOBUF, STRING
  * Importance: high

`key.converter.reference.subject.name.strategy`
: Set the subject reference name strategy for key. 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.
  <br/>
  * Type: string
  * Default: DefaultReferenceSubjectNameStrategy
  * Importance: high

### How should we connect to your data?

`name`
: Sets a name for your connector.
  <br/>
  * Type: string
  * Valid Values: A string at most 64 characters long
  * Importance: high

### 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.
  <br/>
  * 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.
  <br/>
  * 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.
  <br/>
  * Type: string
  * Importance: high

`kafka.api.secret`
: Secret associated with Kafka API key. Required when kafka.auth.mode==KAFKA_API_KEY.
  <br/>
  * Type: password
  * Importance: high

### Authentication method

`authentication.method`
: Select how you want to authenticate with Snowflake. Supported options are Password and OAuth 2.0.
  <br/>
  Password (default): Uses private key authentication.
  <br/>
  OAuth 2.0 (Bring your own app): Uses Snowflake OAuth authentication.
  <br/>
  * Type: string
  * Default: Password
  * Importance: high

`secret.manager.enabled`
: Fetch sensitive configuration values from a secret manager.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: high

### Secret manager configuration

`secret.manager`
: Select the secret manager to use for retrieving sensitive data.
  <br/>
  * Type: string
  * Importance: high

`secret.manager.managed.configs`
: Select the configurations to fetch their values from the secret manager.
  <br/>
  * Type: list
  * Importance: high

`secret.manager.provider.integration.id`
: Select an existing provider integration that has access to your secret manager.
  <br/>
  * Type: string
  * Importance: high

### How should we connect to your Snowflake database?

`snowflake.url.name`
: The URL for accessing your Snowflake account, in the form of [https:/](https:/)/<account_name>.<region_id>.snowflakecomputing.com:443. Note that the [https://](https://) and port number are optional. The region ID is not used if your account is in the AWS US West region and you are not using AWS PrivateLink.
  <br/>
  * Type: string
  * Importance: high

`snowflake.user.name`
: User login name for the Snowflake account
  <br/>
  * Type: string
  * Importance: high

`snowflake.role.name`
: Access control role to use when inserting the rows into the table. If Ingestion method is Snowpipe_Streaming Ingestion then it’s required. If Ingestion method is Snowpipe then it’s not required and use default role
  <br/>
  * Type: string
  * Importance: low

`snowflake.database.name`
: The name of the database that contains the table to insert rows into.
  <br/>
  * Type: string
  * Importance: high

`snowflake.private.key`
: The private key to authenticate the user. Include only the key, not the header or footer. If the key is split across multiple lines, remove the line breaks. You can provide either an unencrypted key or an encrypted key. If you use an encrypted key, provide the snowflake.private.key.passphrase parameter so Snowflake can decrypt the key. Use this parameter only if the snowflake.private.key parameter value is encrypted.
  <br/>
  * Type: password
  * Importance: high

`snowflake.useHttpsProxy`
: Enable HTTPS proxy for Snowflake connections
  <br/>
  * Type: boolean
  * Default: false
  * Importance: low

`snowflake.https.proxyHost`
: HTTPS proxy host for Snowflake connections
  <br/>
  * Type: string
  * Importance: low

`snowflake.https.proxyPort`
: HTTPS proxy port for Snowflake connections
  <br/>
  * Type: string
  * Importance: low

`snowflake.https.nonProxyHosts`
: Hosts to bypass HTTPS proxy for Snowflake connections
  <br/>
  * Type: string
  * Importance: low

`snowflake.https.proxyUser`
: HTTPS proxy username for Snowflake connections
  <br/>
  * Type: string
  * Importance: low

`snowflake.https.proxyPassword`
: HTTPS proxy password for Snowflake connections
  <br/>
  * Type: password
  * Importance: low

`snowflake.disable.ocsp.checks`
: If set to true, disables OCSP (Online Certificate Status Protocol) certificate revocation checks for Snowflake JDBC connections. This should only be enabled in environments where OCSP servers are not accessible. Default is false.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: low

### Database details

`snowflake.schema.name`
: The name of the schema that contains the table to insert rows into.
  <br/>
  * Type: string
  * Importance: high

`snowflake.topic2table.map`
: Map of topics to tables (optional). Format : comma-separated tuples, e.g. <topic-1>:<table-1>,<topic-2>:<table-2>,…
  <br/>
  * Type: string
  * Importance: high

### Snowflake connection

`snowflake.ingestion.method`
: Choose the preferred ingestion method. The connector supports the SNOWPIPE (default) and SNOWPIPE_STREAMING for Kafka data ingestion. Using SNOWPIPE_STREAMING may provide a cost-benefit for your Snowflake project.
  <br/>
  * Type: string
  * Default: SNOWPIPE
  * Valid Values: snowflake.ingestion.method must be SNOWPIPE_STREAMING for using OAuth Authentication Method.
  * Importance: high

### Connection details

`snowflake.metadata.createtime`
: If the value is set to FALSE, the CreateTime property value is omitted from the metadata in the RECORD_METADATA column. The default value is TRUE.
  <br/>
  * Type: boolean
  * Default: true
  * Importance: medium

`snowflake.private.key.passphrase`
: If snowflake.private.key is encrypted, this passphrase is used to decrypt the key. If the value of this parameter is not empty, Kafka uses this phrase to try to decrypt the private key.
  <br/>
  * Type: password
  * Default: [hidden]
  * Importance: medium

`snowflake.metadata.topic`
: If the value is set to FALSE, the topic property value is omitted from the metadata in the RECORD_METADATA column. The default value is TRUE.
  <br/>
  * Type: boolean
  * Default: true
  * Importance: medium

`snowflake.metadata.offset.and.partition`
: If the value is set to FALSE, the Offset and Partition property values are omitted from the metadata in the RECORD_METADATA column. The default value is TRUE.
  <br/>
  * Type: boolean
  * Default: true
  * Importance: medium

`snowflake.streaming.metadata.connectorPushTime`
: If the value is set to TRUE, the ConnectorPushTime property value is added to the metadata in the RECORD_METADATA column. The default value is FALSE. This works with only SNOWPIPE_STREAMING ingestion mode.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: medium

`oauth.client.id`
: Client (or Application) ID of your Snowflake Security Integration
  <br/>
  * Type: string
  * Importance: medium

`snowflake.metadata.all`
: If the value is set to FALSE, the metadata in the RECORD_METADATA column is completely empty. The default value is TRUE.
  <br/>
  * Type: boolean
  * Default: true
  * Importance: medium

`oauth.client.secret`
: Client secret of your Snowflake Security Integration.
  <br/>
  * Type: password
  * Importance: high

`snowflake.enable.schematization`
: Specify to TRUE to enable schema detection and evolution for Kafka Connector with Snowpipe Streaming. The default value is FALSE
  <br/>
  * Type: boolean
  * Default: false
  * Importance: medium

`oauth.refresh.token`
: OAuth 2.0 Refresh Token for Snowflake.
  <br/>
  * Type: password
  * Importance: high

`buffer.flush.time`
: Number of seconds between buffer flushes, where the flush is from the Kafka’s memory cache to the internal stage. The default value is 120 seconds. Minimum value allowed is 10 for snowflake.ingestion.method=SNOWPIPE, and 1 for snowflake.ingestion.method=SNOWPIPE_STREAMING. The connector uses buffer.count.records and buffer.size.bytes=10,000,000 (10MB) as well. Whichever comes first, the connector will flush Kafka records to Snowflake.
  <br/>
  * Type: long
  * Default: 120
  * Valid Values: Value must be greater than 10 in SNOWPIPE method OR greater than 1 in SNOWPIPE_STREAMING method
  * Importance: low

`oauth.auth.endpoint`
: Auth endpoint - {snowflake.url.name}/oauth/authorize
  <br/>
  * Type: string
  * Default: “”
  * Importance: high

`buffer.count.records`
: Number of records between buffer flushes, where the flush is from the Kafka’s memory cache to the internal stage. The default and minimum value is 10,000 records. The connector uses buffer.flush.time and buffer.size.bytes=10,000,000 (10MB) as well. Whichever comes first, the connector will flush Kafka records to Snowflake.
  <br/>
  * Type: long
  * Default: 10000
  * Valid Values: [10000,…]
  * Importance: low

`oauth.token.endpoint`
: Token endpoint - {snowflake.url.name}/oauth/token-request
  <br/>
  * Type: string
  * Default: “”
  * Importance: high

`buffer.size.bytes`
: Kafka records are cached in a buffer (per partition) before being written to Snowflake as data files. The buffer size defaults to 10000000 bytes (10 MB). The records are compressed when written to Snowflake. Because of the compression, the size of the cached records buffer may be larger that the size of the resulting data files created in Snowflake.
  <br/>
  * Type: long
  * Default: 10000000
  * Valid Values: [10000000,…,100000000]
  * Importance: low

`oauth.query.params`
: In the form of scope=session:role:{snowflakeRoleName}
  <br/>
  * Type: string
  * Default: “”
  * Importance: high

`snowflake.streaming.iceberg.enabled`
: Interested in integrating with Iceberg? Take a look at Tableflow — a solution designed to simplify and streamline your Iceberg workflows, refer [https://www.confluent.io/product/tableflow/](https://www.confluent.io/product/tableflow/).
  <br/>
  This property specifies whether the connector ingests data into an Iceberg table. The connector fails if this property doesn’t match the actual table type, refer [https://docs.snowflake.com/en/user-guide/kafka-connector-iceberg](https://docs.snowflake.com/en/user-guide/kafka-connector-iceberg).
  <br/>
  * Type: boolean
  * Default: false
  * Importance: medium

`enable.task.fail.on.authorization.errors`
: If set to true, the connector task will fail whenever it encounters an authorization error from Snowflake.
  <br/>
  * Type: boolean
  * Default: true
  * Importance: low

`enable.dynamic.flush`
: If set to true, each connector task can consume from up to 2x topic partitions depending on the buffer size.
  <br/>
  This option is useful for workloads with many topic partitions but low throughput per partition.
  <br/>
  Note: Enabling this configuration may degrade performance. Because each task handles more partitions, flush operations may be triggered more frequently with smaller batches of data, based on the connector’s buffer footprint.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: low

`snowflake.streaming.channel.name.include.connector.name`
: If enabled, the Snowflake Streaming channel names are prefixed with the connector name. This config enables/disables usage of channel names that were used solely in Kafka Connector versions 2.1.0 and 2.1.1 and is intended for users that previously used these versions and did not update the connector since. IMPORTANT: Enabling this config in any other case will result in data duplication. This configuration is only applicable when snowflake.ingestion.method is set to SNOWPIPE_STREAMING.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: low

`snowflake.streaming.enable.handling.infinity.values`
: Whether to enable correct handling for infinity and negative infinity values in float/double fields when using Snowpipe Streaming. When enabled, infinity values will be correctly ingested as ‘Inf’ or ‘-Inf’. When disabled, infinity values will be treated as ‘NaN’ for backward compatibility. This configuration is only applicable when snowflake.ingestion.method is set to SNOWPIPE_STREAMING.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: low

`snowflake.streaming.max.client.lag`
: Decides how often the buffer in the Ingest SDK will be flushed. The default value is 30 seconds. Setting this value too low may degrade performance, while setting it too high may increase latency. This configuration is only applicable when snowflake.ingestion.method is set to SNOWPIPE_STREAMING.
  <br/>
  * Type: int
  * Default: 30
  * Valid Values: [1,…,300]
  * Importance: low

### Error handling

`errors.tolerance`
: Use this property if you would like to configure the connector’s error handling behavior differently from the Connect framework’s.
  <br/>
  * Type: string
  * Default: all
  * Importance: low

### Consumer configuration

`max.poll.interval.ms`
: The maximum delay between subsequent consume requests to Kafka. This configuration property may be used to improve the performance of the connector, if the connector cannot send records to the sink system. Defaults to 300000 milliseconds (5 minutes).
  <br/>
  * Type: long
  * Default: 300000 (5 minutes)
  * Valid Values: [60000,…,1800000] for non-dedicated clusters and [60000,…] for dedicated clusters
  * Importance: low

`max.poll.records`
: The maximum number of records to consume from Kafka in a single request. This configuration property may be used to improve the performance of the connector, if the connector cannot send records to the sink system. Defaults to 500 records.
  <br/>
  * Type: long
  * Default: 500
  * Valid Values: [1,…,500] for non-dedicated clusters and [1,…] for dedicated clusters
  * Importance: low

### Number of tasks for this connector

`tasks.max`
: The number of tasks for the connector. Each task is limited to a number of topic partitions based on the buffer.size.bytes configuration, e.g., 10 MB -> 50 Topic Partitions, 20 MB-> 25 Topic Partitions, 50 MB -> 10 Topic Partitions, and 100 MB -> 5 Topic Partitions.
  <br/>
  * Type: int
  * Valid Values: [1,…]
  * Importance: high

### Additional Configs

`consumer.override.auto.offset.reset`
: Defines the behavior of the consumer when there is no committed position (which occurs when the group is first initialized) or when an offset is out of range. You can choose either to reset the position to the “earliest” offset (the default) or the “latest” offset. You can also select “none” if you would rather set the initial offset yourself and you are willing to handle out of range errors manually. More details: [https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#auto-offset-reset](https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#auto-offset-reset)
  <br/>
  * Type: string
  * Importance: low

`consumer.override.isolation.level`
: Controls how to read messages written transactionally. If set to read_committed, consumer.poll() will only return transactional messages which have been committed. If set to read_uncommitted (the default), consumer.poll() will return all messages, even transactional messages which have been aborted. Non-transactional messages will be returned unconditionally in either mode.  More details: [https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#isolation-level](https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#isolation-level)
  <br/>
  * Type: string
  * Importance: low

`header.converter`
: The converter class for the headers. This is used to serialize and deserialize the headers of the messages.
  <br/>
  * Type: string
  * Importance: low

`key.converter.use.schema.guid`
: The schema GUID to use for deserialization when using ConfigSchemaIdDeserializer. This allows you to specify a fixed schema GUID to be used for deserializing message keys. Only applicable when key.converter.key.schema.id.deserializer is set to ConfigSchemaIdDeserializer.
  <br/>
  * Type: string
  * Importance: low

`key.converter.use.schema.id`
: The schema ID to use for deserialization when using ConfigSchemaIdDeserializer. This allows you to specify a fixed schema ID to be used for deserializing message keys. Only applicable when key.converter.key.schema.id.deserializer is set to ConfigSchemaIdDeserializer.
  <br/>
  * Type: int
  * 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.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.auto.register.schemas`
: Specify if the Serializer should attempt to register the Schema.
  <br/>
  * 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.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.enhanced.avro.schema.support`
: Enable enhanced schema support to preserve package information and Enums. Applicable for Avro Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.enhanced.protobuf.schema.support`
: Enable enhanced schema support to preserve package information. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.flatten.unions`
: Whether to flatten unions (oneofs). Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.generate.index.for.unions`
: Whether to generate an index suffix for unions. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.generate.struct.for.nulls`
: Whether to generate a struct variable for null values. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.int.for.enums`
: Whether to represent enums as integers. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.latest.compatibility.strict`
: Verify latest subject version is backward compatible when use.latest.version is true.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.object.additional.properties`
: Whether to allow additional properties for object schemas. Applicable for JSON_SR Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.optional.for.nullables`
: Whether nullable fields should be specified with an optional label. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.optional.for.proto2`
: Whether proto2 optionals are supported. Applicable for Protobuf Converters.
  <br/>
  * 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.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.use.latest.version`
: Use latest version of schema in subject for serialization when auto.register.schemas is false.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.use.optional.for.nonrequired`
: Whether to set non-required properties to be optional. Applicable for JSON_SR Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.use.schema.guid`
: The schema GUID to use for deserialization when using ConfigSchemaIdDeserializer. This allows you to specify a fixed schema GUID to be used for deserializing message values. Only applicable when value.converter.value.schema.id.deserializer is set to ConfigSchemaIdDeserializer.
  <br/>
  * Type: string
  * Importance: low

`value.converter.use.schema.id`
: The schema ID to use for deserialization when using ConfigSchemaIdDeserializer. This allows you to specify a fixed schema ID to be used for deserializing message values. Only applicable when value.converter.value.schema.id.deserializer is set to ConfigSchemaIdDeserializer.
  <br/>
  * Type: int
  * Importance: low

`value.converter.wrapper.for.nullables`
: Whether nullable fields should use primitive wrapper messages. Applicable for Protobuf Converters.
  <br/>
  * 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.
  <br/>
  * Type: boolean
  * Importance: low

`behavior.on.null.values`
: Specify how the connector should handle tombstone records. A tombstone record is defined as a record where the entire value field is null.
  <br/>
  This property supports the following values:
  <br/>
  - DEFAULT: When the Kafka connector encounters a tombstone record, it inserts an empty JSON string in the content column.
  - IGNORE (default): The Kafka connector skips tombstone records and does not insert rows for these records.
  <br/>
  * Type: string
  * Default: IGNORE
  * Importance: low

`key.converter.key.schema.id.deserializer`
: The class name of the schema ID deserializer for keys. This is used to deserialize schema IDs from the message headers.
  <br/>
  * Type: string
  * Default: io.confluent.kafka.serializers.schema.id.DualSchemaIdDeserializer
  * Importance: low

`key.converter.key.subject.name.strategy`
: How to construct the subject name for key schema registration.
  <br/>
  * 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:
  <br/>
  BASE64 to serialize DECIMAL logical types as base64 encoded binary data and
  <br/>
  NUMERIC to serialize Connect DECIMAL logical type values in JSON/JSON_SR as a number representing the decimal value.
  <br/>
  * Type: string
  * Default: BASE64
  * Importance: low

`value.converter.flatten.singleton.unions`
: Whether to flatten singleton unions. Applicable for Avro and JSON_SR Converters.
  <br/>
  * 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.
  <br/>
  * 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.
  <br/>
  * Type: string
  * Default: DefaultReferenceSubjectNameStrategy
  * Importance: low

`value.converter.replace.null.with.default`
: Whether to replace fields that have a default value and that are null to the default value. When set to true, the default value is used, otherwise null is used. Applicable for JSON Converter.
  <br/>
  * Type: boolean
  * Default: true
  * Importance: low

`value.converter.schemas.enable`
: Include schemas within each of the serialized values. Input messages must contain schema and payload fields and may not contain additional fields. For plain JSON data, set this to false. Applicable for JSON Converter.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: low

`value.converter.value.schema.id.deserializer`
: The class name of the schema ID deserializer for values. This is used to deserialize schema IDs from the message headers.
  <br/>
  * Type: string
  * Default: io.confluent.kafka.serializers.schema.id.DualSchemaIdDeserializer
  * 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.
  <br/>
  * Type: string
  * Default: TopicNameStrategy
  * Importance: low

### Auto-restart policy

`auto.restart.on.user.error`
: Enable connector to automatically restart on user-actionable errors.
  <br/>
  * Type: boolean
  * Default: true
  * Importance: medium

## Troubleshooting

For Snowflake troubleshooting, see [Troubleshooting Issues](https://docs.snowflake.com/en/user-guide/kafka-connector-ts.html#troubleshooting-issues) in the Snowflake documentation.

<a id="cc-snowflake-oauth-token-expiration"></a>

### OAuth Refresh Token Expiration

When you use OAuth 2.0 authentication, the connector uses a refresh token to
automatically obtain new access tokens. However, the refresh token itself has a
TTL that is configured in your Snowflake security integration
using the `OAUTH_REFRESH_TOKEN_VALIDITY` parameter.

**Error symptoms**

When the refresh token expires, the connector will fail with an error similar
to:

```text
[SF_KAFKA_CONNECTOR] Retry count:0 caught an exception for
operation:FETCH_OAUTH_TOKEN with message:[SF_KAFKA_CONNECTOR] Exception:
Fetching OAuth token fail Error Code: 1004 Detail: Fail to get OAuth token
from authorization server
```

**Resolution steps**

To resolve this issue and refresh the OAuth token:

1. Navigate to the Cloud Console and open your connector configuration.
2. Click **Edit configuration** to enter edit mode.
3. In the **Authentication** section, click the **Connect with Snowflake**
   button.
4. You will be redirected to Snowflake to re-authenticate. Enter your
   credentials when prompted.
5. After successful authentication, you will be redirected back to Confluent Cloud, and the connector configuration automatically updates with a new refresh token.
6. Save the connector configuration.
7. The connector should now resume normal operation.

**Preventing token expiration**

To minimize interruptions due to token expiration:

* Set the `OAUTH_REFRESH_TOKEN_VALIDITY` to a longer period when creating your
  security integration (up to 90 days / 7776000 seconds).
* Monitor the connector status regularly and refresh the token before it
  expires.
* For more information on token validity, see [Snowflake OAuth FAQ](https://community.snowflake.com/s/article/FAQs-Snowflake-OAuth).

#### NOTE
As of June 2024, the minimum TTL for refresh tokens is 1 hour (3600
seconds) and the maximum is 90 days (7776000 seconds). However, these values
may change. Always refer to the latest Snowflake documentation for current
limits.

### Insufficient permission error

The Snowflake user lacks the privileges required to operate on the `BASE` schema.
Ensure the user has the necessary permissions to perform operations on this schema.

**Error symptoms**

When Snowflake Sink connector tasks fail due to insufficient privileges,
the error message indicates that the connector lacks the permissions required to operate.

```text
[SF_KAFKA_CONNECTOR] Exception: Failed to create pipe
Error Code: 2009
Detail: Failed to create pipe on Snowflake, please check that you have permission to do so.
Message: SQL access control error: Insufficient privileges to operate on schema 'BASE'
```

**Resolution steps**

To resolve this issue, verify and grant the required permissions for
the Snowflake user on the `BASE` schema and its associated tables.

For more information on role and privilege configuration, see the following resources:

- [Creating a role to use the Kafka connector](https://docs.snowflake.com/en/user-guide/kafka-connector-install#creating-a-role-to-use-the-kafka-connector).
- [Configuring user privileges](#cc-snowflake-db-sink-gen-key-pair).

<a id="cc-snowflake-schema-evolution"></a>

## Schema Evolution Guidelines with Snowpipe Streaming

To enable automatic schema detection and evolution, set `snowflake.enable.schematization=true`. This requires:

1. The role specified in `snowflake.role.name` must have `EVOLVE SCHEMA`
   privilege (or higher, like `OWNERSHIP` or `ALL`) on the target table.

   For Apache Iceberg™ tables, the `EVOLVE SCHEMA` privilege alone is insufficient.
   The connector’s role must be the direct owner of the Iceberg table. This means
   that the `OWNERSHIP` privilege must be assigned directly to the connector role,
   rather than inherited through a Snowflake database role.
2. The target table must have `enable_schema_evolution = Y`

See [Snowflake Schema Evolution documentation](https://docs.snowflake.com/en/user-guide/data-load-schema-evolution) for more details.

### Error Handling

If either requirement is not met, the connector’s behavior depends on the `errors.tolerance` setting:

* `errors.tolerance=NONE`: Connector task fails immediately
* `errors.tolerance=ALL`: Messages are sent to DLQ with error message:
  ```none
  The given row cannot be converted to the internal format: Extra columns: [<column_list>].
  Columns not present in the table shouldn't be specified
  ```

### Recovery Steps

To reprocess failed messages:

1. Fix the schema issues or grant required privileges
2. [Reset the connector’s offsets](https://docs.confluent.io/cloud/current/connectors/offsets.html)

   #### NOTE
   If you reset the connector’s offset when using [Snowpipe Streaming for Kafka](https://docs.snowflake.com/en/user-guide/data-load-snowpipe-streaming-kafka), you may also need to
   reset the Snowflake offset tokens. For more information, see [Reset Offset in Streaming Mode](#custom-offsets-snowpipe-streaming-proc).
3. Restart processing

## Reset Offset with Snowpipe Streaming

<a id="custom-offsets-snowpipe-streaming-proc"></a>

### Reset Offset in Streaming Mode

To reset custom offsets when using Snowpipe Streaming, follow these steps:

1. Stop the connector.
2. Reset the Snowflake offset token to `-1` by executing the following statement in your Snowflake worksheet.
   ```sql
   SELECT SYSTEM$SNOWPIPE_STREAMING_UPDATE_CHANNEL_OFFSET_TOKEN('db.schema.table', 'channel_name', '-1');
   ```

   For more information on usage , see [SYSTEM$SNOWPIPE_STREAMING_UPDATE_CHANNEL_OFFSET_TOKEN](https://docs.snowflake.com/en/sql-reference/functions/system_snowpipe_streaming_update_channel_offset_token).

   #### NOTE
   If your topic has multiple partitions, repeat these steps for each corresponding channel.
3. Reset your offsets in Confluent Cloud.
4. Restart the connector.

#### WARNING
After this action, if any previously ingested offsets are sent to Snowflake, they will be processed. This introduces the risk of duplicates.

## Suggested Reading

The following blog post provides an introduction to the Snowflake Sink connector and a scenario walkthrough.

Blog post: [Announcing the Snowflake Sink connector for Apache Kafka in Confluent Cloud](https://www.confluent.io/blog/kafka-snowflake-connector-now-available-in-confluent-cloud/)

## Frequently asked questions

Find answers to frequently asked questions about the Snowflake Sink connector for Confluent Cloud.

### Why does my connector fail with `Invalid encrypted private key` errors?

This error occurs when the private key format is not compatible with the connector’s encryption requirements.

Common causes and solutions:

* **Non-FIPS 140-2 compliant encryption**: The connector only supports private keys encrypted using FIPS 140-2 compliant algorithms. If your private key is encrypted with a passphrase, verify the encryption algorithm is FIPS 140-2 compliant, or use an unencrypted private key. Regenerate your private key using the [recommended steps](#cc-snowflake-db-sink-gen-key-pair) which use FIPS-compliant encryption.
* **Incorrect key format**: Ensure you provide only the key content between `--BEGIN RSA PRIVATE KEY--` and `--END RSA PRIVATE KEY--` as a single line without headers or footers.

### Why does my connector fail with `JWT_TOKEN_INVALID_USER_IN_ISSUER` errors?

This error occurs when the username specified in the JWT issuer does not exist in the Snowflake account. The Confluent Cloud Console often displays this as a generic `Cannot connect to Snowflake, due to JWT token is invalid` validation error on the `snowflake.url.name`, `snowflake.user.name`, and `snowflake.private.key` fields.

Common causes and solutions:

* **Username does not match LOGIN_NAME**: The user name configured for `snowflake.user.name` must match the `LOGIN_NAME` of the Snowflake user, not its `NAME`. These values can differ. Run the [DESCRIBE USER](https://docs.snowflake.com/en/sql-reference/sql/desc-user) command in Snowflake and verify that `LOGIN_NAME` matches the configured username.
* **Invalid characters in the username**: Remove any extra characters, such as a stray backslash or slash, from `snowflake.user.name`.

For more information, see [JWT_TOKEN_INVALID_USER_IN_ISSUER](https://docs.snowflake.com/en/user-guide/key-pair-auth-troubleshooting#jwt-token-invalid-user-in-issuer) in the Snowflake documentation.

### Why is my connector stuck in `PROVISIONING` state?

If your connector remains in `PROVISIONING` state indefinitely without logs or errors, this indicates validation or authentication problems during initial setup.

Common causes and solutions:

* **Invalid Kafka API credentials**: The connector cannot validate the Kafka API key and secret. Verify your `kafka.api.key` and `kafka.api.secret` are correct and have permissions to access the specified topics. To verify permissions, check the API key’s ACLs in the Confluent Cloud Console under **Data Integration > API Keys**, or use the Confluent CLI command `confluent kafka acl list --api-key <key>` to confirm the key has read access to the topics.
* **Invalid Snowflake credentials**: Verify your Snowflake URL, username, and private key are correctly configured. Test your Snowflake credentials using a Snowflake client.
* **Network connectivity**: Ensure the connector can reach both your Kafka cluster and Snowflake instance, including for cross-region or cross-cloud deployments.
* **Topic validation failure**: The connector validates that specified topics exist and are accessible. Verify all topics in your configuration exist and your Kafka API key has permissions to read from them.

If the connector remains stuck after verifying these settings, delete and recreate the connector with the corrected configuration.

### Why does my connector fail with `Insufficient privileges` errors?

Permission errors occur when the Snowflake user lacks the required privileges to perform operations on the database, schema, or tables.

Common causes and solutions:

* **Missing table creation privileges**: The user needs `CREATE TABLE` privilege on the schema. Grant this privilege using `GRANT CREATE TABLE ON SCHEMA <schema_name> TO ROLE <role_name>;`
* **Missing pipe creation privileges**: The user needs `CREATE PIPE` privilege when using the Snowpipe ingestion method. Grant using `GRANT CREATE PIPE ON SCHEMA <schema_name> TO ROLE <role_name>;`
* **Missing stage privileges**: The user needs `CREATE STAGE` and `USAGE` privileges on stages. Grant using `GRANT CREATE STAGE ON SCHEMA <schema_name> TO ROLE <role_name>;`
* **Missing EVOLVE SCHEMA privilege**: When using `snowflake.enable.schematization=true`, the role needs `EVOLVE SCHEMA` privilege on the target table. Grant using `GRANT EVOLVE SCHEMA ON TABLE <table_name> TO ROLE <role_name>;`

For complete privilege requirements, see [Creating a role to use the Kafka connector](https://docs.snowflake.com/en/user-guide/kafka-connector-install#creating-a-role-to-use-the-kafka-connector) and [Configuring user privileges](#cc-snowflake-db-sink-gen-key-pair).

### Why does my connector fail with `Incompatible table` schema errors?

This error occurs when the connector attempts to write data to a Snowflake table with an incompatible schema:

```none
[SF_KAFKA_CONNECTOR] Exception: Incompatible table Error Code: 5003
Detail: Table doesn't have a compatible schema
```

Common causes and solutions:

* **NOT NULL constraints on additional columns**: If your Snowflake table has columns with NOT NULL constraints that are not present in the Kafka message schema, the insert will fail. Either remove the NOT NULL constraint or ensure your Kafka messages include all required fields.
* **Schema evolution disabled**: When `snowflake.enable.schematization=false`, the connector can’t automatically add new columns. To enable schema evolution with Snowpipe Streaming mode, set `snowflake.enable.schematization=true` and ensure the role has `EVOLVE SCHEMA` privilege on the target table. If you create the table manually, you must set `enable_schema_evolution = Y` on the table. Schema evolution is only supported with Snowpipe Streaming mode.
* **Data type mismatches**: Verify that Kafka message field types are compatible with the Snowflake table column types.

To enable automatic schema evolution when using Snowpipe Streaming mode, set `snowflake.enable.schematization=true` and if you create the table manually, ensure your target table has `enable_schema_evolution = Y`. For more information, see [Schema Evolution Guidelines with Snowpipe Streaming](#cc-snowflake-schema-evolution).

### Why is data not appearing in my Snowflake table?

The connector shows `RUNNING` status but data is not visible in the Snowflake table.

Common causes and solutions:

* **Snowpipe load failures**: When using the Snowpipe ingestion method, the connector can successfully write data files to the Snowflake stage, but the Snowpipe load process can fail to ingest those files into the target table. Check the Snowflake [COPY_HISTORY](https://docs.snowflake.com/en/sql-reference/functions/copy_history.html) view to identify failed file loads and error details. You can also verify staged files using `LIST @<database_name>.<schema_name>.%<table_name>;` to confirm files are present but not loaded.
* **Data not flushed yet**: Records are buffered before being flushed to Snowflake. When using the Snowpipe ingestion method, data is flushed by default every 120 seconds or when 10,000 records accumulate. Wait for the flush interval to elapse or reduce `buffer.flush.time` for testing.
* **Empty messages or tombstones**: Verify your Kafka topic contains actual data and not just tombstone messages. Check the Kafka topic in the Confluent Cloud Console to confirm message content.
* **Stage file visibility**: Check the internal stage in Snowflake to verify data files are being created. Run `LIST @<database_name>.<schema_name>.%<table_name>;` to see staged files.

For more information on the Snowflake Sink connector workflow, see [Workflow for the Kafka Connector](https://docs.snowflake.com/en/user-guide/kafka-connector-overview.html#workflow-for-the-kafka-connector).

### Why does my connector fail with JDBC communication errors?

Connection failures occur when the connector cannot communicate with Snowflake:

```none
Cannot connect to Snowflake, due to JDBC driver encountered communication error.
Message: HTTP status=503.
```

Common causes and solutions:

* **Snowflake service unavailable**: Snowflake may be experiencing temporary issues. Check the [Snowflake Status Page](https://status.snowflake.com/) for service incidents.
* **Network connectivity**: Verify network connectivity between Confluent Cloud and Snowflake. For private networking, ensure VPC peering or PrivateLink is correctly configured.
* **Proxy configuration**: If using a proxy, verify all proxy settings are correct including `snowflake.https.proxyHost`, `snowflake.https.proxyPort`, and proxy credentials.
* **Port 80 blocked**: OCSP certificate checks require port 80 to be open. Ensure your network allows outbound traffic on ports 80 and 443. For more information, see the [prerequisites](#cc-snowflake-db-sink-prereqs).

Review the connector logs in the Confluent Cloud Console for detailed error messages. If the issue persists, verify your Snowflake URL format follows the pattern `https://<account_locator>.<region_id>.<cloud_provider>.snowflakecomputing.com:443`.

For more information about specific JDBC error codes and their resolutions, see [JDBC Errors](https://docs.snowflake.com/en/user-guide/client-connectivity-troubleshooting/error-messages#jdbc-errors) in the Snowflake documentation.

### How do I determine the correct number of tasks for my connector?

Task count directly impacts connector performance and is limited by the buffer size and number of topic partitions.

The `buffer.size.bytes` property determines the maximum number of topic partitions a single task can handle:

* **10 MB buffer**: Up to 50 topic partitions per task
* **20 MB buffer**: Up to 25 topic partitions per task
* **50 MB buffer**: Up to 10 topic partitions per task
* **100 MB buffer**: Up to 5 topic partitions per task

To calculate the required number of tasks:

1. Count the total number of partitions across all topics the connector consumes.
2. Divide by the partition limit for your buffer size.
3. Round up to get the minimum number of tasks.

For example, if you have 200 topic partitions and use the default 10 MB buffer allowing 50 partitions per task, you need at least 4 tasks.

If your connector experiences lag or slow data ingestion, increase the task count to distribute partitions across more tasks. Alternatively, you can enable the `enable.dynamic.flush` feature, which allows each task to consume from up to twice the normal topic partition limit. Enabling this feature comes with a performance tradeoff.

### Why does my OAuth authentication keep failing after several days?

OAuth refresh token expiration causes authentication failures after the token’s TTL expires.

The refresh token has a Time-To-Live based on the `OAUTH_REFRESH_TOKEN_VALIDITY` setting in your Snowflake security integration. The minimum TTL is 1 hour and the maximum is 90 days.

For resolution steps, see [OAuth Refresh Token Expiration](#cc-snowflake-oauth-token-expiration).

To prevent future expirations:

* Set `OAUTH_REFRESH_TOKEN_VALIDITY` to the maximum value of 90 days when creating your security integration.
* Monitor connector status regularly and refresh the token before it expires.
* Consider using private key authentication if OAuth token management is challenging for your use case.

### Why are my table names different from my topic names?

The connector automatically adjusts Kafka topic names that do not conform to Snowflake table naming requirements.

Snowflake has specific [Identifier Requirements](https://docs.snowflake.com/en/sql-reference/identifiers-syntax.html#identifier-requirements) for table names. When a topic name violates these requirements, the connector converts it to a safe name:

* Special characters such as hyphens are replaced with underscores.
* A hash suffix is appended to prevent naming conflicts.

For example:

* Topic `my-topic-name` becomes table `my_topic_name_021342`
* Topic `my_topic_name` becomes table `my_topic_name` without changes

To avoid unexpected table names:

* Use topic names that conform to Snowflake identifier requirements.
* Use alphanumeric characters and underscores only.
* Avoid special characters, spaces, or hyphens in topic names.

For more information, see [Target table naming guidelines](#cc-snowflake-table-suffixes).

## 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)