fix: Fix potential deadlock in Snowflake destination (#5553)

hermanschaaf · web-flow · commit ebf0eb82a0f2 · 2022-12-13T08:01:35.000Z
(Similar to: #5550) If a worker returns an error, it will no longer be consuming from its `writeChan`, and in this case we should exit the loop early to avoid deadlock.
diff --git a/plugins/destination/snowflake/client/write.go b/plugins/destination/snowflake/client/write.go
@@ -91,7 +91,7 @@ func (c *Client) writeResource(ctx context.Context, table *schema.Table, resourc
 }
 
 func (c *Client) Write(ctx context.Context, tables schema.Tables, res <-chan *plugins.ClientResource) error {
-	eg := errgroup.Group{}
+	eg, gctx := errgroup.WithContext(ctx)
 	workers := make(map[string]*worker, len(tables))
 
 	if _, err := c.db.ExecContext(ctx, createOrReplaceFileFormat); err != nil {
@@ -109,12 +109,22 @@ func (c *Client) Write(ctx context.Context, tables schema.Tables, res <-chan *pl
 			writeChan: writeChan,
 		}
 		eg.Go(func() error {
-			return c.writeResource(ctx, t, writeChan)
+			return c.writeResource(gctx, t, writeChan)
 		})
 	}
 
-	for r := range res {
-		workers[r.TableName].writeChan <- r.Data
+	done := false
+	for !done {
+		select {
+		case r, ok := <-res:
+			if !ok {
+				done = true
+				break
+			}
+			workers[r.TableName].writeChan <- r.Data
+		case <-gctx.Done():
+			done = true
+		}
 	}
 	for _, w := range workers {
 		close(w.writeChan)
