Sitemap
Open in app

Sign in

Medium Logo
Write
Search

Sign in

When Your Snowflake Kafka Sink Connector Writes Nothing: A Step–By–Step Debug Diary (and the Fix!)

Adrian Lee Xinhan
7 min readAug 7, 2025

--

Press enter or click to view image in full size

Overview:

I was working with a client and what begins as a standard data pipeline — ingesting JSON messages from a Confluent Cloud Kafka topic into a Snowflake table — can sometimes turn into a complex troubleshooting odyssey.

To give some background, we have set up a Snowflake Kafka Sink Connector on Kafka Connect standalone (version 3.7.2) to ingest keyed JSON messages from a compacted topic (prodtopic) in Confluent Cloud into a Snowflake table (for example KAFKA_DEMO.PUBLIC.RAW_ORDERS). Despite the connector running and no explicit errors, no data appears in Snowflake.

This blog will tell you how to perform debugging step by step.

Environment:

  • Kafka Connect version: 3.7.2 (standalone mode)
  • Kafka topic: prodtopic
  • Snowflake table: RAW_ORDERS with VARIANT columns for metadata and content
  CREATE OR REPLACE TABLE KAFKA_DEMO.PUBLIC.RAW_ORDERS (
       RECORD_METADATA            VARIANT,
       RECORD_CONTENT             VARIANT
   );
  • Snowflake user: <YOUR USERNAME> with INSERT and USAGE privileges for Snowpipe and also the PUBLIC key attached
  • Kafka cluster: Confluent Cloud (ap-southeast-1)

Symptom

  • Stage and Snowpipe were created automatically.
  • Connector log showed HTTP 200 from Snowflake Ingest API.
  • No rows ever appeared in RAW_ORDERS table
  • No error logs inside the Kafka Connect Logs
  • The repeated Bootstrap broker ... disconnected and Cancelled in-flight API_VERSIONS request warnings are the key indicators. This means your connector task is failing at the most basic step: talking to the Kafka cluster. The Calling PUT with 0 records log is a symptom of this; since the consumer can't connect to fetch messages, the batch to send to Snowflake is always empty.

Connector log snippet:

[2025-07-31 12:21:58,206] INFO [snowflake-sink-connector-fresh|task-0] [Consumer clientId=connector-consumer-snowflake-sink-connector-fresh-0, groupId=connect-snowflake-sink-connector-fresh] Cancelled in-flight API_VERSIONS request with correlation id 1193 due to node -1 being disconnected (elapsed time since creation: 5183ms, elapsed time since send: 5183ms, request timeout: 30000ms) (org.apache.kafka.clients.NetworkClient:353)

[2025-07-31 12:21:58,207] WARN [snowflake-sink-connector-fresh|task-0] [Consumer clientId=connector-consumer-snowflake-sink-connector-fresh-0, groupId=connect-snowflake-sink-connector-fresh] Bootstrap broker pkc-xxxxx.ap-southeast-1.aws.confluent.cloud:9092 (id: -1 rack: null) disconnected

From the above, the root cause seems to be a network connectivity failure between your Kafka Connect worker and your Confluent Cloud brokers. The connector cannot establish a stable connection to Kafka, so it never gets a chance to fetch any records. However, let’s troubleshoot this to understand it better.

Confirm Connection to Confluent Cloud endpoint

Our first priority is to check the connection between the machine running Kafka Connect and the Confluent Cloud endpoint (pkc-xxxx.ap-southeast-1.aws.confluent.cloud:9092).

  • Firewall/Security Groups: Ensure that the outbound traffic on port 9092 is allowed from the server hosting your Kafka Connect worker. If you’re on a cloud provider like AWS, check the instance’s Security Group outbound rules and any Network ACLs.
  • DNS Resolution: Confirm that the Connect worker’s machine can correctly resolve the Confluent Cloud broker hostname. You can test this with a command like:
nslookup pkc-xxxx.ap-southeast-1.aws.confluent.cloud

From the above, it was able to perform the nslookup and confirms the DNS resolution is not the problem. This means our VM was able to correctly find the IP address for the Confluent Cloud broker.

Confirm if there was any firewall or network security rule blocking the connection

Since nslookup works, the next step is to test the actual port connection. Run kafkacat from the same machine as your Kafka Connect worker. This will prove if the port is open and accessible. We will need to provide our Confluent Cloud API key and secret.

# Replace with your actual credentials and topic name
kcat -b pkc-xxxx.ap-southeast-1.aws.confluent.cloud:9092 \
     -X security.protocol=SASL_SSL \
     -X sasl.mechanisms=PLAIN \
     -X sasl.username="<Your-Confluent-API-Key>" \
     -X sasl.password="<Your-Confluent-API-Secret>" \
     -t prodtopic -C -c 1

If all is correct, and if we get an output from the topic, then it proves that network connectivity is working correctly.

From the above, the root cause is therefore not a firewall. The issue is now likely to be in the our Connector file properties. The connector’s internal Kafka client is not configured correctly to talk to Confluent Cloud, which is why it was failing while kcat succeeded.

Let us now check the Connector file properties.

Checking our Connector file properties

With us eliminating the connectivity issue, this now boils down to 2 aspects

  1. Kafka Connect standalone properties file (connect-standalone.properties)
  2. Snowflake Sink Connector file (connect-snowflake-sink.properties)

Let us first check our Kafka Connect standalone properties.

bootstrap.servers=pkc-xxxx.ap-southeast-1.aws.confluent.cloud:9092
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="PKXHL7C6LT3WZRSH" password="rRnkIjJb7h73EhOwrGuDFuvw/s7N1fx/7DSLRgYhlnvAvxW1UqUqIg83uHQiAfGI";

plugin.path=/opt/kafka-3.7.2/plugins

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true

offset.storage.file.filename=/opt/kafka-3.7.2/connect.offsets
offset.flush.interval.ms=10000

Let us try and connect to the properties files and produce a message

/opt/kafka-3.7.2/bin/kafka-console-producer.sh \
  --bootstrap-server pkc-xxxx.ap-southeast-1.aws.confluent.cloud:9092 \
  --topic prodtopic \
  --producer.config /path/to/your/client.properties \
  --property parse.key=true \
  --property key.separator=:

After you run the command, your terminal will show a > prompt and wait. Now, you can type or paste your message and press Enter to send it.

order-5001:{"customer_id":"CUST-1138","item":"Laptop Stand","quantity":1,"amount":45.50,"currency":"SGD"}
Press enter or click to view image in full size
Press enter or click to view image in full size

From the above, we now know there is nothing wrong with the Kafka Connect Standalone properties file. The only path left is our Snowflake-sink connector properties file.

We need to ensure out Snowflake Sink file configuration includes two critical things: the correct security properties to connect to Confluent Cloud and the correct converters to read your JSON data.

  • Converters: Our connector doesn’t know how to read JSON messages without being told. It receives raw bytes from Kafka and, without a converter, discards them, resulting in “0 records” being processed.
  • Security Properties: Our kcat command used SASL_SSL to authenticate. Your connector must be configured with the same settings.

The deafult Kafka Connect configuration looks something like this below

name=snowflake-sink-connector-fresh

connector.class=com.snowflake.kafka.connector.SnowflakeSinkConnector

tasks.max=1
topics=prodtopic

snowflake.url.name=XXXX.snowflakecomputing.com:443
snowflake.user.name=ADRIAN
snowflake.private.key=xxxx
snowflake.database.name=KAFKA_DEMO
snowflake.schema.name=PUBLIC
snowflake.topic2table.map=prodtopic:RAW_ORDERS

bootstrap.servers=pkc-XXXX.ap-southeast-1.aws.confluent.cloud:9092

kafka.auth.mode=KAFKA_API_KEY
kafka.api.key=XXXXXXX
kafka.api.secret=XXXXXX
key.converter=org.apache.kafka.connect.json.JsonConverter

value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=true
input.data.format=JSON

key.converter.schemas.enable=false
value.converter.schemas.enable=false

consumer.auto.offset.reset=earliest

For the above, there are several items and conflicting properties in this configuration that will prevent it from working. The main issue is that we are using non-standard properties for authenticating with Confluent Cloud. The Kafka Connect framework will ignore them, causing the connection to fail.

  • We need to update our connector configuration file with the correct security.protocol, sasl.jaas.config, sasl.mechanism
  • We need to explicitly tell Kafka Connect to force these settings onto the internal consumer. You do this by adding the prefix consumer.override. to all of your Kafka client properties. Let’s update our snowflake-sink.properties file to match this.
name=snowflake-sink-connector-fresh
connector.class=com.snowflake.kafka.connector.SnowflakeSinkConnector
tasks.max=1
topics=prodtopic

# -- Snowflake Connection --
snowflake.url.name=XXXX.snowflakecomputing.com
snowflake.user.name=ADRIAN
snowflake.private.key=...your...single...line...private...key...
snowflake.database.name=KAFKA_DEMO
snowflake.schema.name=PUBLIC
snowflake.warehouse.name=DEMO_BUILD_WH
snowflake.topic2table.map=prodtopic:RAW_ORDERS

# -- Confluent Cloud Connection with Consumer Overrides --
consumer.override.bootstrap.servers=pkc-xxxx.ap-southeast-1.aws.confluent.cloud:9092
consumer.override.security.protocol=SASL_SSL
consumer.override.sasl.mechanism=PLAIN
consumer.override.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="PKXHL7C6LT3WZRSH" password="rRnkIjJb7h73EhOwrGuDFuvw/s7N1fx/7DSLRgYhlnvAvxW1UqUqIg83uHQiAfGI";

# -- Data Converters --
key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

# -- Consumer Behavior --
consumer.override.auto.offset.reset=earliest

If all is correct, we can

  1. Start the connector from your /opt/kafka-3.7.2/ directory:
./bin/connect-standalone.sh config/connect-standalone.properties config/snowflake-sink.properties

2. Produce a test message to your Kafka topic.

# In /opt/kafka-3.7.2/config
/opt/kafka-3.7.2/bin/kafka-console-producer.sh \
  --bootstrap-server pkc-xxx.ap-southeast-1.aws.confluent.cloud:9092 \
  --topic prodtopic \
  --producer.config config/client.properties \
  --property parse.key=true \
  --property key.separator=:
>103:{"customer_name":"Final Test","amount":999}

Or you can go to the Confluent console to produce the message

Press enter or click to view image in full size

3. Check Snowflake after 1–2 minutes. This time, it should work.

-- This should now show a pipe created by the connector
show pipes in schema KAFKA_DEMO.PUBLIC;

-- This should now show your test message
select * from KAFKA_DEMO.PUBLIC.RAW_ORDERS;
Press enter or click to view image in full size

Success!

Final Thoughts

The Snowflake Kafka Sink connector is beautifully boring when everything lines up—files appear, pipes fire, and data lands within seconds. But one mismatch in table schema or privileges can leave you staring at an empty table for hours. Next time it happens, follow the breadcrumbs…and you’ll fix it in minutes instead of half a day.

Key takeways

  • Incorrect Kafka Authentication: The connector’s internal consumer wasn’t inheriting credentials. Fix: Prefix all Kafka client properties with consumer.override..
  • Invalid Snowflake Private Key: The key was in a multi-line PEM format. Fix: Convert the key to a single-line string with no headers/footers.
  • Missing Snowflake Warehouse: The connector needs a warehouse to create its objects. Fix: Explicitly set the snowflake.warehouse.name property.
  • A Rare Snowflake URL Bug: Using an organization-based URL caused the Snowpipe API to fail silently. Fix: Switch to your account locator URL.
  • “Poison Pill” Messages: A single malformed JSON message on the topic crashed the connector task. Fix: Set errors.tolerance = all to skip bad records.
  • Ingestion Latency: Data took minutes to appear due to default settings. Fix: Lower buffer.flush.time and buffer.count.records for faster feedback during testing.

Stay curious and happy streaming!

--

--

Adrian Lee Xinhan
Adrian Lee Xinhan

Written by Adrian Lee Xinhan

355 followers
79 following

Views are personal and don't represent my employer

Help

Status

About

Careers

Press

Blog

Store

Privacy

Rules

Terms

Text to speech