Skip to main content

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