Snowflake Connector for Kafka: Install and configure¶

Download the Kafka connector files¶

Download the Kafka connector JAR file from either of the following locations:

Confluent Hub:

https://www.confluent.io/hub/

The package includes all dependencies required to use either an encrypted or unencrypted private key for key pair authentication. For more information, see Using key pair authentication and key rotation later in this topic.

Maven Central Repository:

https://mvnrepository.com/artifact/com.snowflake

When using this version you need to download the Bouncy Castle cryptography libraries (JAR files):

• https://mvnrepository.com/artifact/org.bouncycastle/bc-fips/2.1.0

• https://mvnrepository.com/artifact/org.bouncycastle/bcpkix-fips/2.1.8

Download these files to the same local folder as the Kafka connector JAR file.

The source code for the connector is available at https://github.com/snowflakedb/snowflake-kafka-connector.

Install the Kafka connector¶

Install the Kafka connector using the instructions provided for installing other connectors:

https://docs.confluent.io/current/connect/userguide.html

Install Apache Kafka¶

1. Download the Kafka package from the Kafka official website.

2. In a terminal window, change to the directory where you downloaded the package file.

3. Run the following command to decompress the kafka_<scala_version>-<kafka_version>.tgz file:

   CopyExpand

   tar xzvf kafka_<scala_version>-<kafka_version>.tgz
   

   Show lessSee more

   Scroll to top

Install the JDK¶

Install and configure the Java Development Kit (JDK) version 11 or higher. Snowflake tests with the Standard Edition (SE) of the JDK. The Enterprise Edition (EE) is expected to be compatible but has not been tested.

If you have previously installed the JDK, you can skip this section.

1. Download the JDK from the Oracle JDK website.

2. Install or decompress the JDK.

3. Following the instructions for your operating system, set the environment variable JAVA_HOME to point to the directory containing the JDK.

Download the Kafka connector JAR files¶

1. Download the Kafka connector JAR file from the Maven Central Repository:

   https://mvnrepository.com/artifact/com.snowflake

2. Download the Bouncy Castle cryptography library jar files:

   • https://mvnrepository.com/artifact/org.bouncycastle/bc-fips/2.1.0

   • https://mvnrepository.com/artifact/org.bouncycastle/bcpkix-fips/2.1.8

3. If your Kafka data is streamed in Apache Avro format, download the Avro JAR file (1.11.4):

   • https://mvnrepository.com/artifact/org.apache.avro/avro

The source code for the connector is available at https://github.com/snowflakedb/snowflake-kafka-connector.

Install the Kafka connector¶

Copy the JAR files you downloaded in Installing the connector for open source Apache Kafka to the <kafka_dir>/libs folder.

Schematization and validation properties¶

These properties control how the connector processes and validates data. For new installations, the defaults work well. If you’re migrating from v3, review Migrate from Kafka connector v3 to v4 for guidance on which values to use.

snowflake.enable.schematization

Controls whether incoming records are schematized into individual table columns or wrapped into legacy VARIANT columns.

When true (the default), record fields are mapped to individual table columns by name. When false, the connector stores records in two VARIANT columns (RECORD_CONTENT and RECORD_METADATA), matching v3 behavior.

Default:

true

snowflake.validation

Controls where data validation and schema evolution are performed.

server_side (default): Validation is performed by the Snowflake backend, aligned with COPY and Snowpipe behavior. Invalid records are captured in Error Tables. Supports both default pipe and user-defined pipe modes.

client_side: The connector validates data types and schema compatibility before sending rows to Snowflake. Supports Dead Letter Queue (DLQ) for invalid records. Only works with the default pipe mode.

For details, see Validation and error handling.

Default:

server_side

Migration and compatibility properties¶

These properties are relevant when migrating from v3. For fresh installations, you can skip these and set snowflake.streaming.validate.compatibility.with.classic=false.

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

Controls how auto-generated table names are derived from topic names.

When false (the default), topic names are used as-is for table names, preserving case and special characters. Table names are created as quoted identifiers.

When true, invalid Snowflake identifier characters are replaced with underscores, names are uppercased, and a hash code is appended for uniqueness. This matches v3 behavior.

Default:

false

snowflake.compatibility.enable.column.identifier.normalization

Controls how column identifiers are handled.

When false (the default), column identifiers preserve case and special characters as-is.

When true, column identifiers are normalized to uppercase, matching v3 behavior.

Default:

false

snowflake.streaming.validate.compatibility.with.classic

Enables startup validation that checks whether all migration-related configs are explicitly set. When true, the connector fails at startup with a descriptive error if any of the following configs are missing or incompatible with v3 behavior:

• snowflake.validation must be client_side

• snowflake.compatibility.enable.column.identifier.normalization must be true

• snowflake.compatibility.enable.autogenerated.table.name.sanitization must be true

• snowflake.enable.schematization must be explicitly set to true or false (the default changed between v3 and v4)

• snowflake.streaming.classic.offset.migration must be explicitly set

• snowflake.streaming.classic.offset.migration.include.connector.name must be explicitly set (when offset migration is strict or best_effort)

This prevents accidentally running v4 with a copied v3 configuration without reviewing the breaking changes. Set to false to skip this check after you’ve confirmed your configuration.

Note

For fresh installations (not migrating from v3), set this to false. The compatibility validator is only needed when upgrading from an existing v3 deployment.

For more information, see Migrate from Kafka connector v3 to v4.

Default:

true

Offset migration properties¶

These properties control how v4 migrates committed offsets from v3 Snowpipe Streaming (SSv1) channels. They’re only relevant when migrating from a v3 connector that used snowflake.ingestion.method=SNOWPIPE_STREAMING. If you’re migrating from v3 Snowpipe mode (file-based ingestion), set snowflake.streaming.classic.offset.migration to skip.

snowflake.streaming.classic.offset.migration

Controls how v4 migrates offsets from v3 Snowpipe Streaming (SSv1) channels.

strict: v4 looks up the committed offset from the v3 SSv1 channel and resumes from that point. If the SSv1 channel isn’t found, the connector fails with an error.

best_effort: v4 attempts to look up the committed offset from the v3 SSv1 channel. If the channel isn’t found, v4 falls back to the Kafka consumer group offset.

skip (default): No SSv1 offset migration is performed. v4 uses the Kafka consumer group offset. Use this when migrating from v3 Snowpipe mode (not Snowpipe Streaming).

Default:

skip

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

Controls whether the SSv1 channel name lookup includes the connector name. This must match how your v3 connector was configured. In v3, the snowflake.streaming.channel.name.include.connector.name property controlled whether the connector name was included in the channel name.

Set to true if your v3 connector had snowflake.streaming.channel.name.include.connector.name=true, or if you were running Kafka connector version 2.1.0 or 2.1.1 (those versions included the connector name by default). Set to false otherwise.

Only required when snowflake.streaming.classic.offset.migration is strict or best_effort.

Default:

none (must be explicitly set when offset migration is active)

Error handling properties¶

errors.tolerance

Controls how the connector responds to errors during ingestion.

none (default): The connector task fails on the first error. With server-side validation, error detection is asynchronous, so a few records after the broken one may still be ingested before the task fails.

all: The connector continues ingesting data. With client-side validation, invalid records are routed to the DLQ (if configured) or silently dropped.

Warning

Setting errors.tolerance=all without configuring a DLQ topic causes invalid records to be silently dropped when using client-side validation. This can cause data loss.

Default:

none

errors.deadletterqueue.topic.name

The Kafka topic name for the Dead Letter Queue. Only effective when snowflake.validation=client_side and errors.tolerance=all.

Default:

empty (DLQ disabled)

errors.log.enable

When true, errors are logged with details of the failed operation and record properties.

Default:

false

enable.task.fail.on.authorization.errors

When true, the connector task fails immediately on authorization errors from Snowflake. When false, the connector retries.

Default:

false

Caching properties¶

snowflake.cache.table.exists

Enables caching for table existence checks, reducing the number of metadata queries to Snowflake.

Default:

true

snowflake.cache.table.exists.expire.ms

Cache expiration time in milliseconds for table existence checks.

Default:

300000 (5 minutes)

snowflake.cache.pipe.exists

Enables caching for pipe existence checks.

Default:

true

snowflake.cache.pipe.exists.expire.ms

Cache expiration time in milliseconds for pipe existence checks.

Default:

300000 (5 minutes)

Monitoring and diagnostics properties¶

jmx

Enables JMX MBeans for connector metrics. For more information, see Monitor the Snowflake Connector for Kafka.

Default:

true

enable.mdc.logging

Enables MDC (Mapped Diagnostic Context) to prepend connector context to log messages, which is useful when running multiple connector instances.

Default:

false

snowflake.streaming.metadata.connectorPushTime

When true, includes the SnowflakeConnectorPushTime timestamp in RECORD_METADATA. This field records when the connector buffered a record for ingestion and is useful for estimating end-to-end latency.

Default:

true

Advanced properties¶

snowflake.streaming.client.provider.override.map

Overrides for Snowpipe Streaming client properties. Format: key1:value1,key2:value2. Use only after consulting Snowflake Support.

Default:

empty

Other properties¶

snowflake.private.key.passphrase

If the value of this parameter is not empty, the connector uses this phrase to try to decrypt the private key.

tasks.max

Number of tasks, usually the same as the number of CPU cores across the worker nodes in the Kafka Connect cluster. To achieve best performance, Snowflake recommends setting the number of tasks equal to the total number of Kafka partitions, but not exceeding the number of CPU cores. High number of tasks may result in an increased memory consumption and frequent rebalances.

snowflake.topic2table.map

Comma-separated list of topic-to-table mappings in topic:table format. Supports regex patterns for topic names. The regular expressions can’t be ambiguous — any matched topic must match only a single target table.

Both topic and table names can be double-quoted to support special characters (colons, commas, spaces). Unquoted table names are uppercased. Quoted table names preserve case.

For detailed examples including regex patterns, many-to-one mappings, and quoting, see Explicit topic-to-table mapping.

Example:

CopyExpand

snowflake.topic2table.map=topic1:low_range,topic2:low_range,"my:topic":"My_Table"

Show lessSee more

Scroll to top

value.converter.schema.registry.url

If the format is Avro and you are using a Schema Registry Service, this should be the URL of the Schema Registry Service. Otherwise this field should be empty.

value.converter.break.on.schema.registry.error

If loading Avro data from the Schema Registry Service, this property determines if the Kafka connector should stop consuming records if it encounters an error while fetching the schema id. The default value is false. Set the value to true to enable this behavior.

jvm.proxy.host

To enable the Snowflake Kafka Connector to access Snowflake through a proxy server, set this parameter to specify the host of that proxy server.

jvm.proxy.port

To enable the Snowflake Kafka Connector to access Snowflake through a proxy server, set this parameter to specify the port of that proxy server.

jvm.proxy.username

Username that authenticates with the proxy server.

jvm.proxy.password

Password for the username that authenticates with the proxy server.

snowflake.jdbc.map

Example: "snowflake.jdbc.map": "networkTimeout:20,tracing:WARNING"

Additional JDBC properties (see JDBC Driver connection parameter reference) are not validated. These additional properties are not validated, and must not override nor be used instead of required properties such as: jvm.proxy.xxx, snowflake.user.name, snowflake.private.key, snowflake.schema.name, and similar properties.

Specifying either of the following combinations:

• tracing property along with JDBC_TRACE env variable

• database property along with snowflake.database.name

Will result in an ambiguous behavior and the behavior will be determined by the JDBC Driver.

value.converter.basic.auth.credentials.source

If you are using the Avro data format and require secure access to the Kafka schema registry, set this parameter to the string “USER_INFO”, and set the value.converter.basic.auth.user.info parameter described below. Otherwise, omit this parameter.

value.converter.basic.auth.user.info

If you are using the Avro data format and require secure access to the Kafka schema registry, set this parameter to the string “<user_ID>:<password>”, and set the value.converter.basic.auth.credentials.source parameter described above. Otherwise, omit this parameter.

snowflake.metadata.createtime

If 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 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 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.metadata.all

If value is set to FALSE, the metadata in the RECORD_METADATA column is completely empty. The default value is TRUE.

transforms

Specify to skip tombstone records encountered by the Kafka connector and not load them into the target table. A tombstone record is defined as a record where the entire value field is null.

Set the property value to "tombstoneHandlerExample".

Note

Use this property with the Kafka community converters (that is, value.converter property value) only (for example, org.apache.kafka.connect.json.JsonConverter or org.apache.kafka.connect.json.AvroConverter). To manage tombstone record handling with the Snowflake converters, use the behavior.on.null.values property instead.

transforms.tombstoneHandlerExample.type

Required when setting the transforms property.

Set the property value to "io.confluent.connect.transforms.TombstoneHandler"

behavior.on.null.values

Specify how the Kafka connector should handle tombstone records. A tombstone record is defined as a record where the entire value field is null. For Snowpipe, this property is supported by the Kafka connector version 1.5.5 and later. For Snowpipe Streaming, this property is supported by the Kafka connector version 2.1.0 and later.

This property supports the following values:

DEFAULT

When the Kafka connector encounters a tombstone record, it inserts an empty JSON string in the content column.

IGNORE

The Kafka connector skips tombstone records and does not insert rows for these records.

The default value is DEFAULT.

Note

Tombstone records ingestion varies by the ingestion methods:

• For Snowpipe, the Kafka connector uses Snowflake converters only. To manage tombstone record handling with the Kafka community converters, use the transform and transforms.tombstoneHandlerExample.type properties instead.

• For Snowpipe Streaming, the Kafka connector uses community converters only.

Records sent to Kafka brokers must not be NULL because these records will be dropped by the Kafka connector resulting in missing offsets. The missing offsets will break the Kafka connector in specific use cases. It is recommended that you use tombstone records instead of NULL records.

Externalizing secrets¶

Snowflake strongly recommends externalizing secrets such as the private key and storing them in an encrypted form or in a key management service such as AWS Key Management Service (KMS), Microsoft Azure Key Vault, or HashiCorp Vault. This can be accomplished by using a ConfigProvider implementation on your Kafka Connect cluster.

For more information, see the Confluent description of this service.