owner: fix memory accumulated when owner consume etcd update slow (#1224

)
pingcap · Dec 18, 2020 · 04e0284 · 04e0284
1 parent e620b5f
commit 04e0284
Showing 1 changed file with 5 additions and 2 deletions.
diff --git a/cdc/owner.go b/cdc/owner.go
@@ -1095,7 +1095,8 @@ func (o *Owner) watchFeedChange(ctx context.Context) chan struct{} {
 				return
 			default:
 			}
-			wch := o.etcdClient.Client.Watch(ctx, kv.TaskPositionKeyPrefix, clientv3.WithFilterDelete(), clientv3.WithPrefix())
+			cctx, cancel := context.WithCancel(ctx)
+			wch := o.etcdClient.Client.Watch(cctx, kv.TaskPositionKeyPrefix, clientv3.WithFilterDelete(), clientv3.WithPrefix())
 
 			for resp := range wch {
 				if resp.Err() != nil {
@@ -1108,10 +1109,12 @@ func (o *Owner) watchFeedChange(ctx context.Context) chan struct{} {
 				// operations should be resolved in future release.
 
 				select {
-				case <-ctx.Done():
 				case output <- struct{}{}:
+				default:
+					// in case output channel is full, just ignore this event
 				}
 			}
+			cancel()
 		}
 	}()
 	return output