Troubleshooting the Kafka connector¶

Raw bytes¶

The filename <connector_name>/<table_name>/<partition>/offset_(<key>/<value>_)<timestamp>.gz includes the exact offset of the record that was not converted from raw bytes to the source file format. To resolve issues, resend the record to the Kafka connector as a new record.

Source file format (Avro, JSON, or protobuf)¶

If Snowpipe could not load data from files in the internal stage created for the Kafka topic, the Kafka connector moves the files to the stage for the target table in the source file format.

If a set of files has multiple issues, the FIRST_ERROR_MESSAGE column in the COPY_HISTORY output only indicates the first error encountered. To view all errors in the files, it is necessary to retrieve the files from the table stage, upload them to a named stage, and then execute a COPY INTO <table> statement with the VALIDATION_MODE copy option set to RETURN_ALL_ERRORS. The VALIDATION_MODE copy option instructs a COPY statement to validate the data to be loaded and return results based on the validation option specified. No data is loaded when this copy option is specified. In the statement, reference the set of files you had attempted to load using the Kafka connector.

When any issues with the data files are resolved, you can load the data manually using one or more COPY statements.

The following example references data files located in the table stage for the mytable table in the mydb.public database and schema.

To validate data files in the table stage and resolve errors:

1. List all files located in the table stage using LIST.

   For example:

   LIST @mydb.public.%mytable;
   

   Copy

   The examples in this section presume that JSON is the source format for the data files.

2. Download the files created by Kafka connector to your local machine using GET.

   For example, download the files to a directory named data on your local machine:

   Linux or macOS:

   GET @mydb.public.%mytable file:///data/;
   

   Copy

   Microsoft Windows:

   GET @mydb.public.%mytable file://C:\data\;
   

   Copy

3. Create a named internal stage using CREATE STAGE that stores data files with the same format as your source Kafka files.

   For example, create a internal stage named kafka_json that stores JSON files:

   CREATE STAGE kafka_json FILE_FORMAT = (TYPE = JSON);
   

   Copy

4. Upload the files you downloaded from the table stage using PUT.

   For example, upload the files downloaded to the data directory on your local machine:

   Linux or macOS:

   PUT file:///data/ @mydb.public.kafka_json;
   

   Copy

   Microsoft Windows:

   PUT file://C:\data\ @mydb.public.kafka_json;
   

   Copy

5. Create a temporary table with two variant columns for testing purposes. The table is only used to validate staged data file. No data is loaded into the table. The table is dropped automatically when the current user session ends:

   CREATE TEMPORARY TABLE t1 (col1 variant);
   

   Copy

6. Retrieve all errors encountered in the data file by executing a COPY INTO *table* … VALIDATION_MODE = ‘RETURN_ALL_ERRORS’ statement. The statement validates the file in the specified stage. No data is loaded into the table:

   COPY INTO mydb.public.t1
     FROM @mydb.public.kafka_json
     FILE_FORMAT = (TYPE = JSON)
     VALIDATION_MODE = 'RETURN_ALL_ERRORS';
   

   Copy

7. Fix all reported errors in the data files on your local machine.

8. Upload the fixed files to either the table stage or the named internal stage using PUT.

   The following example uploads the files to the table stage, overwriting the existing files:

   Linux or macOS:

   PUT file:///tmp/myfile.csv @mydb.public.%mytable OVERWRITE = TRUE;
   

   Copy

   Windows:

   PUT file://C:\temp\myfile.csv @mydb.public.%mytable OVERWRITE = TRUE;
   

   Copy

9. Load the data into the target table using COPY INTO table without the VALIDATION_MODE option.

   You can optionally use the PURGE = TRUE copy option to delete the data files from the stage once the data is loaded successfully, or manually delete the files from the table stage using REMOVE:

   COPY INTO mydb.public.mytable(RECORD_METADATA, RECORD_CONTENT)
     FROM (SELECT $1:meta, $1:content FROM @mydb.public.%mytable)
     FILE_FORMAT = (TYPE = 'JSON')
     PURGE = TRUE;
   

   Copy