Opening many channels

[jmks]

Hi all

We've been having some instability problems with the connector, but I think we're working through those problems.

One issue has been how long it takes a task to open channels. We have 178 tables with 12 topics, so 2136 channels.

Previously, we observed if it took more than 10 minutes to open the channels, it would have to re-open them, and a task would fall into a loop of opening, closing, re-opening...

We upped the number of tasks (now 8) so that opening 267 channels takes roughly 2 minutes.

But if some tasks fail (for whatever reason), a running task may get 2-3x that many partitions assigned, and end up in the open-then-close cycle again.

Has the channel timeout changed recently? I've observed that channels happen after only 5 minutes now (it seemed longer before but my comparisons are not equal).

Has there been any discussion of opening channels in parallel (as they can be closed in #841)?

[jmks]

Actually, within the same day, opening the channels seems to vary in length by a lot. Enough that the batch after 22:00 appears to timeout:

[2025-02-03 19:13:02,968] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 146323 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 19:13:02,986] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 146246 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 19:13:03,128] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 146393 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 19:13:03,133] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 146409 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 19:13:03,207] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 146595 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 19:13:03,236] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 146593 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 19:13:03,506] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 146895 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 19:13:03,560] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 146897 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 20:03:37,026] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 169379 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 20:03:37,360] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 169888 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 20:03:37,465] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 169977 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 20:07:00,520] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 157728 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 20:07:00,705] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 157870 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 20:07:00,968] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 158143 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 20:07:01,411] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 158778 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 20:07:01,479] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 158740 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 20:07:01,738] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 158808 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 20:07:01,792] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 159136 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 20:07:01,934] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 159089 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 22:03:36,675] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 450898 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 22:03:36,806] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 451009 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 22:03:36,884] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 451085 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 22:03:36,901] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 451110 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 22:03:42,793] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 457000 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 22:03:42,801] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 457145 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 22:03:53,511] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 467707 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 22:03:53,519] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 467761 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 22:47:02,122] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 346536 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 22:47:02,148] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 346505 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 22:47:02,149] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 346509 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 22:47:09,483] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 353917 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 22:47:09,512] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 353937 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 22:47:15,034] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 359425 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 22:47:15,054] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 359428 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)
[2025-02-03 22:47:15,098] INFO [SF_KAFKA_CONNECTOR] task opened with 267 partitions, execution time: 359467 milliseconds (com.snowflake.kafka.connector.SnowflakeSinkTask)

[sfc-gh-mbobowski]

Hi!
Opening channels in parallel is not so easy, because it would require adding some synchronization mechanisms around private final Map<String, TopicPartitionChannel> partitionsToChannel;. Could you try setting ingestion method to SNOWPIPE? It might be a better choice for handling that many partitions.