sink: when context is done, send to chan will be block (pingcap#574)

5kbpers · May 15, 2020 · 0f8a7d2 · 0f8a7d2
1 parent e1951b5
commit 0f8a7d2
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/cdc/sink/mysql.go b/cdc/sink/mysql.go
@@ -399,7 +399,11 @@ func concurrentExec(
 	sendFn := func(rows []*model.RowChangedEvent, keys []string, idx int) {
 		causality.add(keys, idx)
 		jobWg.Add(1)
-		rowsChs[idx] <- rows
+		select {
+		case <-ctx.Done():
+			jobWg.Done()
+		case rowsChs[idx] <- rows:
+		}
 	}
 	for groupKey, multiRows := range rowGroups {
 		for _, rows := range multiRows {