> ## Documentation Index
> Fetch the complete documentation index at: https://docs.automq.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Snowflake Sink Connector

> Configure and operate the Snowflake Sink Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting.

## Overview

The Snowflake Sink Connector continuously writes records from Kafka topics to Snowflake tables. It sits in the data-writing path between Kafka and Snowflake. It uses Snowpipe Streaming to write data from each topic partition to the corresponding Snowflake ingestion channel. It can determine target tables from topic names or route topics to specified tables through mappings.

The Connector can expand structured records into Snowflake columns or write records to the `RECORD_CONTENT` and `RECORD_METADATA` columns. It is suitable for continuously writing business events, logs, and change data.

## Prerequisites

* The target Snowflake database, schema, user, and role must exist and have permission to access the target objects, create or write tables, and use Snowpipe Streaming.
* JWT authentication requires an RSA private key. OAuth authentication requires a client ID, client secret, and the tokens or endpoints required by the selected flow.
* When using an existing table, its structure must be compatible with the record conversion and metadata settings. Managed Iceberg tables also require external volume and catalog configuration.

## License

Uses Apache License 2.0.

## Quick Start

Prepare a Connect Cluster, Kafka topic, and target Snowflake database and schema in advance. Confirm that the Connect Worker can access Kafka and Snowflake and has the required permissions. For creating and managing the Connector, see AutoMQ's [Manage Connectors](../manage-connectors).

```properties theme={null}
connector.class=com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector
tasks.max=1
topics=orders
snowflake.url.name=https://<account-url>
snowflake.user.name=<snowflake-user>
snowflake.private.key=<rsa-private-key>
snowflake.database.name=<database-name>
snowflake.schema.name=<schema-name>
snowflake.role.name=<role-name>
snowflake.authenticator=snowflake_jwt
snowflake.streaming.validate.compatibility.with.classic=false
snowflake.enable.schematization=false
snowflake.autocreate.table.type=snowflake
value.converter=org.apache.kafka.connect.json.JsonConverter
```

Replace the values in angle brackets with actual values. This example subscribes to the `orders` topic, automatically creates the missing Snowflake table in the target schema, and writes records to the `RECORD_CONTENT` and `RECORD_METADATA` columns.

## Configuration

### Connection and Authentication

#### `snowflake.url.name`

Snowflake account URL. This property has no usable runtime default.

* **Type**: `STRING`
* **Default**: None
* **Importance**: High
* **Required**: Yes

#### `snowflake.user.name`

User for JWT or OAuth Snowflake sessions.

* **Type**: `STRING`
* **Default**: None
* **Importance**: High
* **Required**: Yes

#### `snowflake.private.key`

Required by the default `snowflake_jwt` authenticator. A Kafka Config Provider reference can be used.

* **Type**: `PASSWORD`
* **Default**: empty string
* **Importance**: High
* **Required**: Conditionally required

#### `snowflake.private.key.passphrase`

Set only for an encrypted RSA private key. A Config Provider reference can be used.

* **Type**: `PASSWORD`
* **Default**: empty string
* **Importance**: Low
* **Required**: No

#### `snowflake.database.name`

The target Snowflake database must exist and be accessible to the configured role.

* **Type**: `STRING`
* **Default**: None
* **Importance**: High
* **Required**: Yes

#### `snowflake.schema.name`

The target schema must exist and be accessible to the configured role.

* **Type**: `STRING`
* **Default**: None
* **Importance**: High
* **Required**: Yes

#### `snowflake.role.name`

Specifies the Snowflake session role. When OAuth scope inclusion is enabled without an explicit scope, it is also used to derive `session:role:<role>`.

* **Type**: `STRING`
* **Default**: None
* **Importance**: High
* **Required**: Yes

#### `snowflake.authenticator`

Selects JWT or OAuth authentication, which have mutually exclusive credential requirements. Valid values are `snowflake_jwt` and `oauth`.

* **Type**: `STRING`
* **Default**: snowflake\_jwt
* **Importance**: Low
* **Required**: No

#### `snowflake.oauth.client.id`

Required for OAuth. A Config Provider reference defers live credential validation.

* **Type**: `STRING`
* **Default**: empty string
* **Importance**: High
* **Required**: Conditionally required

#### `snowflake.oauth.client.secret`

Required for OAuth. A Config Provider reference defers live credential validation.

* **Type**: `PASSWORD`
* **Default**: empty string
* **Importance**: High
* **Required**: Conditionally required

#### `snowflake.oauth.refresh.token`

An empty value selects the OAuth `client_credentials` flow; a nonempty value selects the refresh-token flow.

* **Type**: `PASSWORD`
* **Default**: empty string
* **Importance**: High
* **Required**: No

#### `snowflake.oauth.token.endpoint`

When unset, the token endpoint is derived from the Snowflake account URL.

* **Type**: `STRING`
* **Default**: None
* **Importance**: High
* **Required**: No

#### `snowflake.oauth.include.scope`

`false` sends no scope. `true` sends the explicit scope, or `session:role:<role>` when no scope is set.

* **Type**: `BOOLEAN`
* **Default**: false
* **Importance**: Low
* **Required**: No

#### `snowflake.oauth.scope`

Used only when `snowflake.oauth.include.scope=true`. An empty value then derives the scope from the configured role.

* **Type**: `STRING`
* **Default**: empty string
* **Importance**: Low
* **Required**: No

### Network Proxy

#### `jvm.proxy.host`

A nonempty value sets process-wide HTTP and HTTPS JVM proxy properties. It must be configured together with `jvm.proxy.port`.

* **Type**: `STRING`
* **Default**: empty string
* **Importance**: Low
* **Required**: Conditionally required

#### `jvm.proxy.port`

Passed as the JVM proxy port string. It must be configured together with `jvm.proxy.host`.

* **Type**: `STRING`
* **Default**: empty string
* **Importance**: Low
* **Required**: Conditionally required

#### `jvm.nonProxy.hosts`

When the proxy host and port are active, this value is appended to the existing `http.nonProxyHosts` JVM property with `|`.

* **Type**: `STRING`
* **Default**: empty string
* **Importance**: Low
* **Required**: No

#### `jvm.proxy.username`

Used only when the proxy host and port are active. It must be configured together with `jvm.proxy.password`.

* **Type**: `STRING`
* **Default**: empty string
* **Importance**: Low
* **Required**: Conditionally required

#### `jvm.proxy.password`

Used only when the proxy host and port are active. It must be configured together with `jvm.proxy.username`.

* **Type**: `PASSWORD`
* **Default**: empty string
* **Importance**: Low
* **Required**: Conditionally required

### Tables and Data Model

#### `snowflake.metadata.all`

Master metadata switch. `false` discards record metadata regardless of the individual field switches. It must remain `true` for managed Iceberg tables.

* **Type**: `BOOLEAN`
* **Default**: true
* **Importance**: Low
* **Required**: No

#### `snowflake.metadata.createtime`

Controls creation-time metadata. It must remain `true` for managed Iceberg tables.

* **Type**: `BOOLEAN`
* **Default**: true
* **Importance**: Low
* **Required**: No

#### `snowflake.metadata.topic`

Controls topic metadata. It must remain `true` for managed Iceberg tables.

* **Type**: `BOOLEAN`
* **Default**: true
* **Importance**: Low
* **Required**: No

#### `snowflake.metadata.offset.and.partition`

Controls both Kafka offset and partition metadata. It must remain `true` for managed Iceberg tables.

* **Type**: `BOOLEAN`
* **Default**: true
* **Importance**: Low
* **Required**: No

#### `snowflake.streaming.metadata.connectorPushTime`

Controls Connector push-time metadata. It must remain `true` for managed Iceberg tables.

* **Type**: `BOOLEAN`
* **Default**: true
* **Importance**: Low
* **Required**: No

#### `snowflake.feature.structured.headers`

`true` preserves converted structured header types; `false` flattens header values to strings. Enabling it can break consumers that depend on the legacy metadata representation.

* **Type**: `BOOLEAN`
* **Default**: false
* **Importance**: Low
* **Required**: No

### Compatibility and Migration

#### `snowflake.streaming.validate.compatibility.with.classic`

The default `true` is a v3 migration safety gate. While enabled, set `snowflake.validation=client_side`, set both `snowflake.compatibility.*` normalization switches to `true`, and explicitly set `snowflake.enable.schematization` and `snowflake.streaming.classic.offset.migration`. When migration mode is `strict` or `best_effort`, also explicitly set `snowflake.streaming.classic.offset.migration.include.connector.name`. For a new v4 Connector that does not require v3 compatibility, explicitly set this property to `false`.

* **Type**: `BOOLEAN`
* **Default**: true
* **Importance**: High
* **Required**: No

### Topic Routing, Validation, and Migration

#### `snowflake.topic2table.map`

Comma-separated `topic:table` mappings. Quoted table names preserve case, while unquoted names are uppercased. An empty value derives table names from topic names.

* **Type**: `STRING`
* **Default**: empty string
* **Importance**: Low
* **Required**: No

#### `snowflake.validation`

Valid values are `server_side` and `client_side`. `server_side` requires error logging on the target table. Managed Iceberg tables do not support `client_side`.

* **Type**: `STRING`
* **Default**: server\_side
* **Importance**: High
* **Required**: No

#### `snowflake.streaming.classic.offset.migration`

Valid values are `skip`, `best_effort`, and `strict`. A v3 Classic channel is consulted only when the SSv2 channel has no committed offset. `strict` fails when the legacy channel is absent, `best_effort` falls back to the Kafka consumer-group offset, and `skip` does not consult legacy channels.

* **Type**: `STRING`
* **Default**: skip
* **Importance**: High
* **Required**: Conditionally required

#### `snowflake.streaming.classic.offset.migration.include.connector.name`

Used only with `strict` or `best_effort`. The value must match whether the v3 Connector included the Connector name in channel names.

* **Type**: `BOOLEAN`
* **Default**: false
* **Importance**: High
* **Required**: Conditionally required

#### `behavior.on.null.values`

Valid values are `default` and `ignore`. `ignore` filters Kafka tombstones. `default` retains the legacy behavior and writes empty JSON content to the target table.

* **Type**: `STRING`
* **Default**: default
* **Importance**: Low
* **Required**: No

### Logging, Metrics, and Advanced Options

#### `jmx`

Enables the Connector's custom Snowflake metrics MBeans.

* **Type**: `BOOLEAN`
* **Default**: true
* **Importance**: High
* **Required**: No

#### `snowflake.streaming.client.provider.override.map`

Advanced Snowpipe Streaming SDK overrides. Use only under guidance from Snowflake Support.

* **Type**: `STRING`
* **Default**: empty string
* **Importance**: Low
* **Required**: No

### Error Handling

#### `errors.tolerance`

Valid values are `all` and `none`. `all` tolerates Connector-side ingestion or validation record failures and can preserve them in a DLQ. `none` fails the Task.

* **Type**: `STRING`
* **Default**: none
* **Importance**: Low
* **Required**: No

#### `errors.log.enable`

Logs tolerated record failures. Consider record sensitivity before enabling detailed framework message logging.

* **Type**: `BOOLEAN`
* **Default**: false
* **Importance**: Low
* **Required**: No

#### `errors.deadletterqueue.topic.name`

Preserves tolerated failures when `errors.tolerance=all`. An empty value disables DLQ output.

* **Type**: `STRING`
* **Default**: empty string
* **Importance**: Low
* **Required**: No

#### `enable.mdc.logging`

Enables global MDC context for Connector logs.

* **Type**: `BOOLEAN`
* **Default**: false
* **Importance**: Low
* **Required**: No

#### `enable.task.fail.on.authorization.errors`

When `true`, a previously observed Snowflake authorization error causes the Task to fail during `preCommit`.

* **Type**: `BOOLEAN`
* **Default**: false
* **Importance**: Low
* **Required**: No

#### `snowflake.compatibility.enable.autogenerated.table.name.sanitization`

`true` sanitizes and uppercases automatically generated table names for v3 compatibility. `false` passes topic names through. Use an explicit `snowflake.topic2table.map` for special names.

* **Type**: `BOOLEAN`
* **Default**: false
* **Importance**: Low
* **Required**: Conditionally required

#### `snowflake.compatibility.enable.column.identifier.normalization`

`true` normalizes column identifiers to uppercase for v3 compatibility.

* **Type**: `BOOLEAN`
* **Default**: false
* **Importance**: Low
* **Required**: Conditionally required

#### `snowflake.enable.schematization`

`true` maps records to individual columns. `false` writes the v3-compatible `RECORD_CONTENT` and `RECORD_METADATA` VARIANT columns.

* **Type**: `BOOLEAN`
* **Default**: true
* **Importance**: Medium
* **Required**: Conditionally required

#### `snowflake.autocreate.table.type`

Valid values are `snowflake`, `iceberg`, and `none`. The first two automatically create a missing table of that type; `none` fails when the table is missing. Existing tables are always used with their current type and schema.

* **Type**: `STRING`
* **Default**: snowflake
* **Importance**: Medium
* **Required**: No

#### `snowflake.iceberg.create.table.options`

SQL clauses appended to the `CREATE` statement for an automatically created managed Iceberg table. Do not include `CATALOG`, `ENABLE_SCHEMA_EVOLUTION`, or `ERROR_LOGGING`. Existing tables ignore this property.

* **Type**: `STRING`
* **Default**: empty string
* **Importance**: Low
* **Required**: No

#### `snowflake.cache.table.exists`

Enables caching of target-table existence checks.

* **Type**: `BOOLEAN`
* **Default**: true
* **Importance**: Low
* **Required**: No

#### `snowflake.cache.table.exists.expire.ms`

Table-existence cache expiration in milliseconds. The minimum value is `1`.

* **Type**: `LONG`
* **Default**: 300000
* **Importance**: Low
* **Required**: No

#### `snowflake.cache.pipe.exists`

Enables caching of Pipe existence checks.

* **Type**: `BOOLEAN`
* **Default**: true
* **Importance**: Low
* **Required**: No

#### `snowflake.cache.pipe.exists.expire.ms`

Pipe-existence cache expiration in milliseconds. The minimum value is `1`.

* **Type**: `LONG`
* **Default**: 300000
* **Importance**: Low
* **Required**: No

#### `snowflake.topic2table.map.regex.replacement`

`true` enables Java regex capture-group replacement in mapped table-name templates. `false` preserves literal `$` characters and the legacy no-substitution behavior.

* **Type**: `BOOLEAN`
* **Default**: false
* **Importance**: Low
* **Required**: No

### Concurrency and Conversion

#### `connector.class`

Use `com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector`. This is a Kafka Connect framework property, not part of the Connector's own `ConfigDef`.

* **Type**: `STRING`
* **Default**: No fixed default
* **Importance**: High
* **Required**: Yes

#### `tasks.max`

Upper bound on the requested Task count. Effective parallelism is also bounded by the assigned topic partitions.

* **Type**: `INT`
* **Default**: 1
* **Importance**: High
* **Required**: No

#### `tasks.max.enforce`

When `true`, Kafka Connect fails a Connector task set that exceeds `tasks.max`.

* **Type**: `BOOLEAN`
* **Default**: true
* **Importance**: Low
* **Required**: No
* **Deprecated**: Yes
* **Replacement**: None

#### `topics`

Comma-separated topic list. Exactly one of `topics` and `topics.regex` must be configured.

* **Type**: `LIST`
* **Default**: empty string
* **Importance**: High
* **Required**: Conditionally required

#### `topics.regex`

Full Java regular expression for topic subscription. Exactly one of `topics` and `topics.regex` must be configured.

* **Type**: `STRING`
* **Default**: empty string
* **Importance**: High
* **Required**: Conditionally required

#### `key.converter`

When unset, inherits the Worker key converter. Converter-specific subproperties depend on the selected plugin and have no universal default.

* **Type**: `CLASS`
* **Default**: `null` (inherits the Worker configuration)
* **Importance**: Low
* **Required**: No

#### `value.converter`

When unset, inherits the Worker value converter. Converter-specific properties such as `schemas.enable` depend on the selected Converter and have no universal default.

* **Type**: `CLASS`
* **Default**: `null` (inherits the Worker configuration)
* **Importance**: Low
* **Required**: No

#### `errors.deadletterqueue.topic.replication.factor`

Used only when Kafka Connect creates a missing DLQ topic. The value must be valid for the number of brokers in the target Kafka cluster.

* **Type**: `SHORT`
* **Default**: 3
* **Importance**: Medium
* **Required**: No

#### `errors.deadletterqueue.context.headers.enable`

Adds `__connect.errors.*` context headers to DLQ records written by the framework.

* **Type**: `BOOLEAN`
* **Default**: false
* **Importance**: Medium
* **Required**: No

## Best Practices

### Route Multiple Topics to Explicit Target Tables

Applicable business scenario: One Connector must write different topics to different Snowflake tables, or a topic name cannot be used directly as a target table name.

Configuration example: Add explicit mappings to the base configuration.

```properties theme={null}
connector.class=com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector
tasks.max=2
topics=orders,customers
snowflake.url.name=https://<account-url>
snowflake.user.name=<snowflake-user>
snowflake.private.key=<rsa-private-key>
snowflake.database.name=<database-name>
snowflake.schema.name=<schema-name>
snowflake.role.name=<role-name>
snowflake.authenticator=snowflake_jwt
snowflake.streaming.validate.compatibility.with.classic=false
snowflake.enable.schematization=false
snowflake.topic2table.map=orders:ORDERS,customers:CUSTOMERS
value.converter=org.apache.kafka.connect.json.JsonConverter
```

Key points: Exact topic matches take precedence, and duplicate or overlapping mappings are rejected. Unquoted table names are converted to uppercase.

### Tolerate Record Errors and Retain Failed Records

Applicable business scenario: Individual records may fail in the Converter or Connector-side processing, but the continuous write task must keep running while failed records are retained for troubleshooting.

Configuration example: Enable Connect error tolerance and a DLQ, and ensure that the DLQ topic is not matched by the current subscription.

```properties theme={null}
connector.class=com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector
tasks.max=1
topics=orders
snowflake.url.name=https://<account-url>
snowflake.user.name=<snowflake-user>
snowflake.private.key=<rsa-private-key>
snowflake.database.name=<database-name>
snowflake.schema.name=<schema-name>
snowflake.role.name=<role-name>
snowflake.authenticator=snowflake_jwt
snowflake.streaming.validate.compatibility.with.classic=false
snowflake.enable.schematization=false
errors.tolerance=all
errors.deadletterqueue.topic.name=orders-dlq
errors.deadletterqueue.context.headers.enable=true
value.converter=org.apache.kafka.connect.json.JsonConverter
```

Key points: The DLQ handles tolerable Kafka Connect and Connector-side record errors; it does not receive records rejected by Snowflake server-side validation. Backend throttling, channel recovery failures, and unrecoverable ingestion errors can still cause the Task to fail. When Kafka Connect creates the DLQ topic, the default replication factor is `3`; if the cluster has fewer brokers, pre-create the topic or explicitly adjust `errors.deadletterqueue.topic.replication.factor`.

## Monitoring

### What to Monitor

Monitor Kafka Connect Worker health, Connector and Task status, throughput, latency, offset commits, errors and retries, as well as Worker JVM heap, GC, threads, and CPU. After enabling error tolerance and a DLQ, also monitor DLQ write volume, failed-record growth, and delivery errors.

### Import the Grafana Dashboard

Download the [AutoMQ Connect Cluster Dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json). In Grafana, select the data source that collects Kafka Connect metrics, confirm that the metric labels match the dashboard variables, and import the JSON file.

## Limitations

* No global record ordering is provided across Tasks, topics, or Kafka partitions.
* The committed Snowflake channel offset plus one is the recovery and Kafka commit boundary for an individual topic partition. If the committed channel status cannot be read, that partition is not reported as safe to commit.
* Replayed records at offsets already known to be committed are skipped, but this mechanism is not an atomic exactly-once guarantee across Tasks, partitions, or tables.
* When `snowflake.autocreate.table.type=none`, a missing target table causes task initialization to fail.
* Managed Iceberg tables are incompatible with `snowflake.validation=client_side`.
* Offset commits and recovery are bounded by topic partitions; atomic checkpoints across multiple partitions or tables are not provided.
* Backend throttling, channel recovery failures, and unrecoverable ingestion errors can cause the task to fail.

## FAQ

### What should I do when the task reports missing required configuration at startup?

Check the Snowflake URL, user, database, schema, and role. The default authenticator is `snowflake_jwt`, which also requires `snowflake.private.key`. When using OAuth, instead check the OAuth client ID and secret.

### Why are records not written to the target table after a topic is configured?

Confirm that only one of `topics` and `topics.regex` is configured, and check whether the subscription matches the actual topic. Then check database, schema, and role permissions, as well as explicit mappings and automatic table-creation settings.

### Why does a Classic compatibility configuration error appear at startup?

The default compatibility validation requires `snowflake.validation=client_side`, both v3 normalization switches set to `true`, and explicit schematization and Classic offset-migration settings. For a new 4.1.0 Connector that does not require v3 compatibility, set `snowflake.streaming.validate.compatibility.with.classic=false`. When migrating from v3, select `strict` or `best_effort` and set `snowflake.streaming.classic.offset.migration.include.connector.name` according to whether the previous Connector enabled `snowflake.streaming.channel.name.include.connector.name`.

### Why cannot structured data be converted?

When schematization is enabled, the value must be convertible to a Map or Struct, and StringConverter or ByteArrayConverter cannot be used. Check the Converter, message structure, and target table permissions.

### Why do failed records not appear in the DLQ?

Confirm that `errors.tolerance=all` is set, the DLQ topic is configured and is not matched by the current subscription, and the Worker provides ErrantRecordReporter and Kafka allows writes to that topic.