Merge pull request #12271 from jingyih/add_watch_notify_interval_flag_in_testing

integration: add WatchProgressNotifyInterval in integration test

clientv3/integration/watch_test.go

@@ -584,6 +584,30 @@ func testWatchWithProgressNotify(t *testing.T, watchOnPut bool) {
 	}
 }
 
+func TestConfigurableWatchProgressNotifyInterval(t *testing.T) {
+	progressInterval := 200 * time.Millisecond
+	clus := integration.NewClusterV3(t,
+		&integration.ClusterConfig{
+			Size:                        3,
+			WatchProgressNotifyInterval: progressInterval,
+		})
+	defer clus.Terminate(t)
+
+	opts := []clientv3.OpOption{clientv3.WithProgressNotify()}
+	rch := clus.RandClient().Watch(context.Background(), "foo", opts...)
+
+	timeout := 1 * time.Second // we expect to receive watch progress notify in 2 * progressInterval,
+	// but for CPU-starved situation it may take longer. So we use 1 second here for timeout.
+	select {
+	case resp := <-rch: // waiting for a watch progress notify response
+		if !resp.IsProgressNotify() {
+			t.Fatalf("expected resp.IsProgressNotify() == true")
+		}
+	case <-time.After(timeout):
+		t.Fatalf("timed out waiting for watch progress notify response in %v", timeout)
+	}
+}
+
 func TestWatchRequestProgress(t *testing.T) {
 	testCases := []struct {
 		name     string

integration/cluster.go

@@ -152,6 +152,8 @@ type ClusterConfig struct {
 
 	EnableLeaseCheckpoint   bool
 	LeaseCheckpointInterval time.Duration
+
+	WatchProgressNotifyInterval time.Duration
 }
 
 type cluster struct {
@@ -296,6 +298,7 @@ func (c *cluster) mustNewMember(t testing.TB) *member {
 			useIP:                       c.cfg.UseIP,
 			enableLeaseCheckpoint:       c.cfg.EnableLeaseCheckpoint,
 			leaseCheckpointInterval:     c.cfg.LeaseCheckpointInterval,
+			WatchProgressNotifyInterval: c.cfg.WatchProgressNotifyInterval,
 		})
 	m.DiscoveryURL = c.cfg.DiscoveryURL
 	if c.cfg.UseGRPC {
@@ -585,6 +588,7 @@ type memberConfig struct {
 	useIP                       bool
 	enableLeaseCheckpoint       bool
 	leaseCheckpointInterval     time.Duration
+	WatchProgressNotifyInterval time.Duration
 }
 
 // mustNewMember return an inited member with the given name. If peerTLS is
@@ -678,6 +682,8 @@ func mustNewMember(t testing.TB, mcfg memberConfig) *member {
 	m.EnableLeaseCheckpoint = mcfg.enableLeaseCheckpoint
 	m.LeaseCheckpointInterval = mcfg.leaseCheckpointInterval
 
+	m.WatchProgressNotifyInterval = mcfg.WatchProgressNotifyInterval
+
 	m.InitialCorruptCheck = true
 
 	lcfg := logutil.DefaultZapLoggerConfig