close watcher

Signed-off-by: lhy1024 <[email protected]>

pkg/election/leadership.go

@@ -196,21 +196,29 @@ func (ls *Leadership) Watch(serverCtx context.Context, revision int64) {
 	defer ticker.Stop()
 	lastReceivedResponseTime := time.Now()
 
-	watcher := clientv3.NewWatcher(ls.client)
-	defer watcher.Close()
-	var watchChanCancel context.CancelFunc
+	var (
+		watcher         clientv3.Watcher
+		watchChanCancel context.CancelFunc
+	)
 	defer func() {
 		if watchChanCancel != nil {
 			watchChanCancel()
 		}
+		if watcher != nil {
+			watcher.Close()
+		}
 	}()
 	for {
 		failpoint.Inject("delayWatcher", nil)
 		if watchChanCancel != nil {
 			watchChanCancel()
 		}
+		if watcher != nil {
+			watcher.Close()
+		}
 		// In order to prevent a watch stream being stuck in a partitioned node,
 		// make sure to wrap context with "WithRequireLeader".
+		watcher = clientv3.NewWatcher(ls.client)
 		watchChanCtx, cancel := context.WithCancel(clientv3.WithRequireLeader(serverCtx))
 		watchChanCancel = cancel
 

pkg/utils/etcdutil/etcdutil.go

@@ -687,21 +687,26 @@ func (lw *LoopWatcher) watch(ctx context.Context, revision int64) (nextRevision
 	defer ticker.Stop()
 	lastReceivedResponseTime := time.Now()
 
-	watcher := clientv3.NewWatcher(lw.client)
-	defer watcher.Close()
+	var watcher clientv3.Watcher
 	var watchChanCancel context.CancelFunc
 	defer func() {
 		if watchChanCancel != nil {
 			watchChanCancel()
 		}
+		if watcher != nil {
+			watcher.Close()
+		}
 	}()
-
 	for {
 		if watchChanCancel != nil {
 			watchChanCancel()
 		}
+		if watcher != nil {
+			watcher.Close()
+		}
 		// In order to prevent a watch stream being stuck in a partitioned node,
 		// make sure to wrap context with "WithRequireLeader".
+		watcher = clientv3.NewWatcher(lw.client)
 		watchChanCtx, cancel := context.WithCancel(clientv3.WithRequireLeader(ctx))
 		watchChanCancel = cancel
 		opts := append(lw.opts, clientv3.WithRev(revision))