etcd_worker: fix missed watch events caused by progress notifications (…

…pingcap#3848)

(cherry picked from commit 06547d9)

pkg/etcd/client.go

@@ -204,7 +204,7 @@ func (c *Client) WatchWithChan(ctx context.Context, outCh chan<- clientv3.WatchR
 			return
 		case response := <-watchCh:
 			lastReceivedResponseTime = c.clock.Now()
-			if response.Err() == nil {
+			if response.Err() == nil && !response.IsProgressNotify() {
 				lastRevision = response.Header.Revision
 			}
 

pkg/orchestrator/etcd_worker_bank_test.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/pingcap/check"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/orchestrator/util"
@@ -121,6 +122,12 @@ func (b *bankReactor) Tick(ctx context.Context, state ReactorState) (nextState R
 
 func (s *etcdWorkerSuite) TestEtcdBank(c *check.C) {
 	defer testleak.AfterTest(c)()
+
+	_ = failpoint.Enable("github.com/pingcap/ticdc/pkg/orchestrator/InjectProgressRequestAfterCommit", "10%return(true)")
+	defer func() {
+		_ = failpoint.Disable("github.com/pingcap/ticdc/pkg/orchestrator/InjectProgressRequestAfterCommit")
+	}()
+
 	totalAccountNumber := 25
 	workerNumber := 10
 	var wg sync.WaitGroup