I have been trying to send data from Kafka to snowflake using the JDBC driver with Kafka connect. some details about the environment: Kafka is running in a Cloudera private cluster (Base 7.1.9). The version of the Snowflake-kafka connector driver is 2.2.2. The configuration of the Kafka (v3.2.0+) connector is: { "buffer.count.records": "2", "buffer.flush.time": "60", "buffer.size.bytes": "5000000", "connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector", "errors.log.enable": "true", "errors.log.include.messages": "true", "key.converter": "org.apache.kafka.connect.storage.StringConverter", "snowflake.database.name": "kafka_db", "snowflake.private.key": "REDACTED", "snowflake.schema.name": "kafka_schema", "snowflake.topic2table.map": "snowflake_kafka_poc:existing_table1", "snowflake.url.name": "https://REDACTED.snowflakecomputing.com", "snowflake.user.name": "kafka_connector_user_1", "tasks.max": "4", "topics": "snowflake_kafka_poc", "value.converter": "com.snowflake.kafka.connector.records.SnowflakeJsonConverter", "secret.properties": "snowflake.private.key", "name": "kafka_snowflake_poc_2" } The error I'm getting in the Kafka log is:     2024-07-11T07:30:53,744 INFO com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] init pipe: SNOWFLAKE_KAFKA_CONNECTOR_kafka_snowflake_poc_2_PIPE_existing_table1_0 2024-07-11T07:30:54,383 INFO com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] Using existing table existing_table1. 2024-07-11T07:30:55,111 INFO com.snowflake.kafka.connector.internal.SnowflakeConnectionServiceV1: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] list stage SNOWFLAKE_KAFKA_CONNECTOR_kafka_snowflake_poc_2_STAGE_existing_table1 retrieved 0 file names 2024-07-11T07:30:55,111 INFO com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] Using existing stage SNOWFLAKE_KAFKA_CONNECTOR_kafka_snowflake_poc_2_STAGE_existing_table1. 2024-07-11T07:30:55,480 INFO com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] pipe SNOWFLAKE_KAFKA_CONNECTOR_kafka_snowflake_poc_2_PIPE_existing_table1_0, recovered from existing pipe 2024-07-11T07:30:56,755 INFO com.snowflake.kafka.connector.internal.SnowflakeConnectionServiceV1: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] list stage SNOWFLAKE_KAFKA_CONNECTOR_kafka_snowflake_poc_2_STAGE_existing_table1 retrieved 0 file names 2024-07-11T07:30:56,758 INFO com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1: [SF_KAFKA_CONNECTOR] pipe SNOWFLAKE_KAFKA_CONNECTOR_kafka_snowflake_poc_2_PIPE_existing_table1_0: cleaner started 2024-07-11T07:30:57,262 WARN com.snowflake.kafka.connector.internal.SnowflakeInternalStage: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] uploadWithoutConnection encountered an exception:JDBC driver encountered IO error. Message: Encountered exception during upload: null. for filePath:kafka_snowflake_poc_2/existing_table1/0/1_1_1720683056775.json.gz in Storage:AZURE 2024-07-11T07:30:57,265 ERROR com.snowflake.kafka.connector.internal.InternalUtils: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] Retry count:0 caught an exception for operation:UPLOAD_FILE_TO_INTERNAL_STAGE_NO_CONNECTION with message:[SF_KAFKA_CONNECTOR] Exception: Failed to execute cached put Error Code: 5018 Detail: Error in cached put command Message: JDBC driver encountered IO error. Message: Encountered exception during upload: null. 2024-07-11T07:30:58,370 WARN com.snowflake.kafka.connector.internal.SnowflakeInternalStage: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] uploadWithoutConnection encountered an exception:JDBC driver encountered IO error. Message: Encountered exception during upload: null. for filePath:kafka_snowflake_poc_2/existing_table1/0/1_1_1720683056775.json.gz in Storage:AZURE 2024-07-11T07:30:58,370 ERROR com.snowflake.kafka.connector.internal.InternalUtils: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] Retry count:1 caught an exception for operation:UPLOAD_FILE_TO_INTERNAL_STAGE_NO_CONNECTION with message:[SF_KAFKA_CONNECTOR] Exception: Failed to execute cached put Error Code: 5018 Detail: Error in cached put command Message: JDBC driver encountered IO error. Message: Encountered exception during upload: null. 2024-07-11T07:31:00,714 WARN com.snowflake.kafka.connector.internal.SnowflakeInternalStage: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] uploadWithoutConnection encountered an exception:JDBC driver encountered IO error. Message: Encountered exception during upload: null. for filePath:kafka_snowflake_poc_2/existing_table1/0/1_1_1720683056775.json.gz in Storage:AZURE 2024-07-11T07:31:00,714 ERROR com.snowflake.kafka.connector.internal.InternalUtils: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] Retry count:2 caught an exception for operation:UPLOAD_FILE_TO_INTERNAL_STAGE_NO_CONNECTION with message:[SF_KAFKA_CONNECTOR] Exception: Failed to execute cached put Error Code: 5018 Detail: Error in cached put command Message: JDBC driver encountered IO error. Message: Encountered exception during upload: null. 2024-07-11T07:31:04,820 WARN com.snowflake.kafka.connector.internal.SnowflakeInternalStage: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] uploadWithoutConnection encountered an exception:JDBC driver encountered IO error. Message: Encountered exception during upload: null. for filePath:kafka_snowflake_poc_2/existing_table1/0/1_1_1720683056775.json.gz in Storage:AZURE 2024-07-11T07:31:04,820 ERROR com.snowflake.kafka.connector.internal.InternalUtils: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] Retry count:4 caught an exception for operation:UPLOAD_FILE_TO_INTERNAL_STAGE_NO_CONNECTION with message:[SF_KAFKA_CONNECTOR] Exception: Failed to execute cached put Error Code: 5018 Detail: Error in cached put command Message: JDBC driver encountered IO error. Message: Encountered exception during upload: null. 2024-07-11T07:31:12,954 WARN com.snowflake.kafka.connector.internal.SnowflakeInternalStage: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] uploadWithoutConnection encountered an exception:JDBC driver encountered IO error. Message: Encountered exception during upload: null. for filePath:kafka_snowflake_poc_2/existing_table1/0/1_1_1720683056775.json.gz in Storage:AZURE 2024-07-11T07:31:12,954 ERROR com.snowflake.kafka.connector.internal.InternalUtils: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] Retry count:8 caught an exception for operation:UPLOAD_FILE_TO_INTERNAL_STAGE_NO_CONNECTION with message:[SF_KAFKA_CONNECTOR] Exception: Failed to execute cached put Error Code: 5018 Detail: Error in cached put command Message: JDBC driver encountered IO error. Message: Encountered exception during upload: null. 2024-07-11T07:31:12,955 ERROR com.snowflake.kafka.connector.internal.SnowflakeConnectionServiceV1: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] Put With Cache(uploadWithoutConnection) failed after multiple retries for stageName:SNOWFLAKE_KAFKA_CONNECTOR_kafka_snowflake_poc_2_STAGE_existing_table1, stageType:AZURE, fullFilePath:kafka_snowflake_poc_2/existing_table1/0/1_1_1720683056775.json.gz 2024-07-11T07:31:12,955 ERROR org.apache.kafka.connect.runtime.WorkerSinkTask: [kafka_snowflake_poc_2|task-0] WorkerSinkTask{id=kafka_snowflake_poc_2-0, context=null} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted. Error: [SF_KAFKA_CONNECTOR] Exception: Failed to upload file with cache Error Code: 2011 Detail: Failed to upload file to Snowflake Stage though credential caching Message: [SF_KAFKA_CONNECTOR] Exception: Max retry exceeded Error Code: 2010 Detail: Api retry exceeded the max retry limit Message: [SF_KAFKA_CONNECTOR] Exception: Failed to execute cached put Error Code: 5018 Detail: Error in cached put command Message: JDBC driver encountered IO error. Message: Encountered exception during upload: null. com.snowflake.kafka.connector.internal.SnowflakeErrors.getException(SnowflakeErrors.java:381) com.snowflake.kafka.connector.internal.SnowflakeErrors.getException(SnowflakeErrors.java:339) com.snowflake.kafka.connector.internal.SnowflakeInternalStage.putWithCache(SnowflakeInternalStage.java:202) com.snowflake.kafka.connector.internal.SnowflakeConnectionServiceV1.lambda$putWithCache$1(SnowflakeConnectionServiceV1.java:855) com.snowflake.kafka.connector.internal.InternalUtils.backoffAndRetry(InternalUtils.java:373) com.snowflake.kafka.connector.internal.SnowflakeConnectionServiceV1.putWithCache(SnowflakeConnectionServiceV1.java:851) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.flush(SnowflakeSinkServiceV1.java:844) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.flushBuffer(SnowflakeSinkServiceV1.java:773) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.access$200(SnowflakeSinkServiceV1.java:373) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1.insert(SnowflakeSinkServiceV1.java:159) com.snowflake.kafka.connector.SnowflakeSinkTask.put(SnowflakeSinkTask.java:304) org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:588) org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336) org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237) org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206) org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:205) org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:260) org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:183) java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) java.base/java.lang.Thread.run(Thread.java:829) com.snowflake.kafka.connector.internal.SnowflakeErrors.getException(SnowflakeErrors.java:381) com.snowflake.kafka.connector.internal.SnowflakeErrors.getException(SnowflakeErrors.java:353) com.snowflake.kafka.connector.internal.InternalUtils.backoffAndRetry(InternalUtils.java:383) com.snowflake.kafka.connector.internal.SnowflakeConnectionServiceV1.putWithCache(SnowflakeConnectionServiceV1.java:851) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.flush(SnowflakeSinkServiceV1.java:844) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.flushBuffer(SnowflakeSinkServiceV1.java:773) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.access$200(SnowflakeSinkServiceV1.java:373) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1.insert(SnowflakeSinkServiceV1.java:159) com.snowflake.kafka.connector.SnowflakeSinkTask.put(SnowflakeSinkTask.java:304) org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:588) org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336) org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237) org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206) org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:205) org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:260) org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:183) java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) java.base/java.lang.Thread.run(Thread.java:829) com.snowflake.kafka.connector.internal.SnowflakeKafkaConnectorException: [SF_KAFKA_CONNECTOR] Exception: Failed to upload file with cache Error Code: 2011 Detail: Failed to upload file to Snowflake Stage though credential caching Message: [SF_KAFKA_CONNECTOR] Exception: Max retry exceeded Error Code: 2010 Detail: Api retry exceeded the max retry limit Message: [SF_KAFKA_CONNECTOR] Exception: Failed to execute cached put Error Code: 5018 Detail: Error in cached put command Message: JDBC driver encountered IO error. Message: Encountered exception during upload: null. com.snowflake.kafka.connector.internal.SnowflakeErrors.getException(SnowflakeErrors.java:381) com.snowflake.kafka.connector.internal.SnowflakeErrors.getException(SnowflakeErrors.java:339) com.snowflake.kafka.connector.internal.SnowflakeInternalStage.putWithCache(SnowflakeInternalStage.java:202) com.snowflake.kafka.connector.internal.SnowflakeConnectionServiceV1.lambda$putWithCache$1(SnowflakeConnectionServiceV1.java:855) com.snowflake.kafka.connector.internal.InternalUtils.backoffAndRetry(InternalUtils.java:373) com.snowflake.kafka.connector.internal.SnowflakeConnectionServiceV1.putWithCache(SnowflakeConnectionServiceV1.java:851) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.flush(SnowflakeSinkServiceV1.java:844) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.flushBuffer(SnowflakeSinkServiceV1.java:773) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.access$200(SnowflakeSinkServiceV1.java:373) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1.insert(SnowflakeSinkServiceV1.java:159) com.snowflake.kafka.connector.SnowflakeSinkTask.put(SnowflakeSinkTask.java:304) org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:588) org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336) org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237) org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206) org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:205) org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:260) org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:183) java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) java.base/java.lang.Thread.run(Thread.java:829) com.snowflake.kafka.connector.internal.SnowflakeErrors.getException(SnowflakeErrors.java:381) com.snowflake.kafka.connector.internal.SnowflakeErrors.getException(SnowflakeErrors.java:353) com.snowflake.kafka.connector.internal.InternalUtils.backoffAndRetry(InternalUtils.java:383) com.snowflake.kafka.connector.internal.SnowflakeConnectionServiceV1.putWithCache(SnowflakeConnectionServiceV1.java:851) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.flush(SnowflakeSinkServiceV1.java:844) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.flushBuffer(SnowflakeSinkServiceV1.java:773) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.access$200(SnowflakeSinkServiceV1.java:373) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1.insert(SnowflakeSinkServiceV1.java:159) com.snowflake.kafka.connector.SnowflakeSinkTask.put(SnowflakeSinkTask.java:304) org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:588) org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336) org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237) org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206) org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:205) org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:260) org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:183) java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) java.base/java.lang.Thread.run(Thread.java:829) at com.snowflake.kafka.connector.internal.SnowflakeErrors.getException(SnowflakeErrors.java:381) ~[snowflake-kafka-connector-2.2.2.jar:?] at com.snowflake.kafka.connector.internal.SnowflakeErrors.getException(SnowflakeErrors.java:353) ~[snowflake-kafka-connector-2.2.2.jar:?] at com.snowflake.kafka.connector.internal.SnowflakeConnectionServiceV1.putWithCache(SnowflakeConnectionServiceV1.java:865) ~[snowflake-kafka-connector-2.2.2.jar:?] at com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.flush(SnowflakeSinkServiceV1.java:844) ~[snowflake-kafka-connector-2.2.2.jar:?] at com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.flushBuffer(SnowflakeSinkServiceV1.java:773) ~[snowflake-kafka-connector-2.2.2.jar:?] at com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.access$200(SnowflakeSinkServiceV1.java:373) ~[snowflake-kafka-connector-2.2.2.jar:?] at com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1.insert(SnowflakeSinkServiceV1.java:159) ~[snowflake-kafka-connector-2.2.2.jar:?] at com.snowflake.kafka.connector.SnowflakeSinkTask.put(SnowflakeSinkTask.java:304) ~[snowflake-kafka-connector-2.2.2.jar:?] at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:588) ~[connect-runtime-3.4.1.7.1.9.4-4.jar:?] at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336) ~[connect-runtime-3.4.1.7.1.9.4-4.jar:?] at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237) ~[connect-runtime-3.4.1.7.1.9.4-4.jar:?] at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206) ~[connect-runtime-3.4.1.7.1.9.4-4.jar:?] at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:205) ~[connect-runtime-3.4.1.7.1.9.4-4.jar:?] at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:260) ~[connect-runtime-3.4.1.7.1.9.4-4.jar:?] at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:183) ~[connect-runtime-3.4.1.7.1.9.4-4.jar:?] at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[?:?] at java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[?:?] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[?:?] at java.lang.Thread.run(Thread.java:829) ~[?:?] 2024-07-11T07:31:12,957 INFO com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] pipe SNOWFLAKE_KAFKA_CONNECTOR_kafka_snowflake_poc_2_PIPE_existing_table1_0: cleaner terminated 2024-07-11T07:31:12,957 INFO com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1: [SF_KAFKA_CONNECTOR] Cleaner terminated by an interrupt: sleep interrupted 2024-07-11T07:31:13,086 INFO com.snowflake.kafka.connector.internal.SnowflakeIngestionServiceV1: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] IngestService Closed 2024-07-11T07:31:13,086 INFO com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] pipe SNOWFLAKE_KAFKA_CONNECTOR_kafka_snowflake_poc_2_PIPE_existing_table1_0: service closed 2024-07-11T07:31:13,088 INFO com.snowflake.kafka.connector.SnowflakeSinkTask: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] task closed, execution time: 0 milliseconds 2024-07-11T07:31:13,088 ERROR org.apache.kafka.connect.runtime.WorkerTask: [kafka_snowflake_poc_2|task-0] WorkerSinkTask{id=kafka_snowflake_poc_2-0, context=null} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception. at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:619) ~[connect-runtime-3.4.1.7.1.9.4-4.jar:?] at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336) ~[connect-runtime-3.4.1.7.1.9.4-4.jar:?] at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237) ~[connect-runtime-3.4.1.7.1.9.4-4.jar:?] at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206) ~[connect-runtime-3.4.1.7.1.9.4-4.jar:?] at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:205) ~[connect-runtime-3.4.1.7.1.9.4-4.jar:?] at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:260) ~[connect-runtime-3.4.1.7.1.9.4-4.jar:?] at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:183) ~[connect-runtime-3.4.1.7.1.9.4-4.jar:?] at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[?:?] at java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[?:?] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[?:?] at java.lang.Thread.run(Thread.java:829) ~[?:?] Caused by: com.snowflake.kafka.connector.internal.SnowflakeKafkaConnectorException: [SF_KAFKA_CONNECTOR] Exception: Failed to upload file with cache Error Code: 2011 Detail: Failed to upload file to Snowflake Stage though credential caching Message: [SF_KAFKA_CONNECTOR] Exception: Max retry exceeded Error Code: 2010 Detail: Api retry exceeded the max retry limit Message: [SF_KAFKA_CONNECTOR] Exception: Failed to execute cached put Error Code: 5018 Detail: Error in cached put command Message: JDBC driver encountered IO error. Message: Encountered exception during upload: null. com.snowflake.kafka.connector.internal.SnowflakeErrors.getException(SnowflakeErrors.java:381) com.snowflake.kafka.connector.internal.SnowflakeErrors.getException(SnowflakeErrors.java:339) com.snowflake.kafka.connector.internal.SnowflakeInternalStage.putWithCache(SnowflakeInternalStage.java:202) com.snowflake.kafka.connector.internal.SnowflakeConnectionServiceV1.lambda$putWithCache$1(SnowflakeConnectionServiceV1.java:855) com.snowflake.kafka.connector.internal.InternalUtils.backoffAndRetry(InternalUtils.java:373) com.snowflake.kafka.connector.internal.SnowflakeConnectionServiceV1.putWithCache(SnowflakeConnectionServiceV1.java:851) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.flush(SnowflakeSinkServiceV1.java:844) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.flushBuffer(SnowflakeSinkServiceV1.java:773) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.access$200(SnowflakeSinkServiceV1.java:373) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1.insert(SnowflakeSinkServiceV1.java:159) com.snowflake.kafka.connector.SnowflakeSinkTask.put(SnowflakeSinkTask.java:304) org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:588) org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336) org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237) org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206) org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:205) org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:260) org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:183) java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) java.base/java.lang.Thread.run(Thread.java:829) com.snowflake.kafka.connector.internal.SnowflakeErrors.getException(SnowflakeErrors.java:381) com.snowflake.kafka.connector.internal.SnowflakeErrors.getException(SnowflakeErrors.java:353) com.snowflake.kafka.connector.internal.InternalUtils.backoffAndRetry(InternalUtils.java:383) com.snowflake.kafka.connector.internal.SnowflakeConnectionServiceV1.putWithCache(SnowflakeConnectionServiceV1.java:851) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.flush(SnowflakeSinkServiceV1.java:844) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.flushBuffer(SnowflakeSinkServiceV1.java:773) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.access$200(SnowflakeSinkServiceV1.java:373) com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1.insert(SnowflakeSinkServiceV1.java:159) com.snowflake.kafka.connector.SnowflakeSinkTask.put(SnowflakeSinkTask.java:304) org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:588) org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336) org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237) org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206) org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:205) org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:260) org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:183) java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) java.base/java.lang.Thread.run(Thread.java:829) at com.snowflake.kafka.connector.internal.SnowflakeErrors.getException(SnowflakeErrors.java:381) ~[snowflake-kafka-connector-2.2.2.jar:?] at com.snowflake.kafka.connector.internal.SnowflakeErrors.getException(SnowflakeErrors.java:353) ~[snowflake-kafka-connector-2.2.2.jar:?] at com.snowflake.kafka.connector.internal.SnowflakeConnectionServiceV1.putWithCache(SnowflakeConnectionServiceV1.java:865) ~[snowflake-kafka-connector-2.2.2.jar:?] at com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.flush(SnowflakeSinkServiceV1.java:844) ~[snowflake-kafka-connector-2.2.2.jar:?] at com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.flushBuffer(SnowflakeSinkServiceV1.java:773) ~[snowflake-kafka-connector-2.2.2.jar:?] at com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1$ServiceContext.access$200(SnowflakeSinkServiceV1.java:373) ~[snowflake-kafka-connector-2.2.2.jar:?] at com.snowflake.kafka.connector.internal.SnowflakeSinkServiceV1.insert(SnowflakeSinkServiceV1.java:159) ~[snowflake-kafka-connector-2.2.2.jar:?] at com.snowflake.kafka.connector.SnowflakeSinkTask.put(SnowflakeSinkTask.java:304) ~[snowflake-kafka-connector-2.2.2.jar:?] at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:588) ~[connect-runtime-3.4.1.7.1.9.4-4.jar:?] ... 11 more 2024-07-11T07:31:13,089 INFO com.snowflake.kafka.connector.SnowflakeSinkTask: [kafka_snowflake_poc_2|task-0] [SF_KAFKA_CONNECTOR] task stopped, total task runtime: 1130759 milliseconds I see in Snowflake the Stage, Pipe and Table being created automatically upon the arrival of data to the Kafka topic. I expect the data from the Kafka topic being uploaded to the Stage in Snowflake, however here is no trace of activity in Snowflake's COPY_HISTORY and no file is uploaded into the Stage. I've followed the instructions shown here: https://docs.snowflake.com/en/user-guide/kafka-connector-installChengwei Semiconductor. While troubleshooting, I've tried to use the information given in the following links reporting similar problems, although none of these are exactly the same shown above: https://community.snowflake.com/s/article/Increasing-the-buffer-size-bytes-in-kafka-snowpipe-mode-results-in-error https://github.com/snowflakedb/snowflake-kafka-connector/issues/525 https://docs.snowflake.com/en/user-guide/client-connectivity-troubleshooting/error-messages (here looking at the JDBC error 10) Also, if I try to create a table beforehand and intentionally give a wrong schema, I get an error of an incompatible schema, so the data should not be an issue. Additionally, I've made sure there is no privilege issue. The user with which the Kafka connector authenticates to Snowflake can upload file to the stage and query the table. Any idea of more things I could try? Any advise is appreciated. ... View more

We are running a sparkly job that runs queries on cloudera CDP hive cluster. The job sometimes stops before a dbwriteTable function, doing nothing and running indefinitely. The job doesn't always stop in the same point, but always during this trywrite function invocation, not catching any error: trywrite = function(sc, new_name, df, log_obj, wait_sec = 600, max_wait = 3600) { start_time = Sys.time() while (difftime(Sys.time(), start_time, units = 'secs') <= max_wait) { print(paste0('Attempt to write table: ', new_name, ' - ', Sys.time())) # Connection is valid? if (!DBI::dbIsValid(sc)) { error(log_obj, paste0('Connection not valid during write table: ', new_name)) stop(paste0('Failed to write table: ', new_name)) } tryCatch({ print(paste0('Writing table: ', new_name)) result = DBI::dbWriteTable(sc, new_name, df) print(paste0('Write completed table: ', new_name, ' - ', Sys.time())) return(result) }, error = function(e) { error(log_obj, paste0('Connection not valid during write table: ', new_name, ' - ', Sys.time())) print(paste0('Error message: ', e$message)) print(paste0('Retrying in', wait_sec, ' seconds: ', Sys.time())) Sys.sleep(wait_sec) }) } stop(paste0('Failed to write table before max time: ', new_name)) }       ... View more