pdClient (ticdc): Enable the pdClient forwarding function to make cdc…

… more stable during network isolation between the PD leader (#11076) (#11086)

close #10849

cdc/api/v2/api_helpers.go

@@ -464,7 +464,8 @@ func (APIV2HelpersImpl) getPDClient(ctx context.Context,
 				},
 				MinConnectTimeout: 3 * time.Second,
 			}),
-		))
+		),
+		pd.WithForwardingOption(config.EnablePDForwarding))
 	if err != nil {
 		return nil, cerror.WrapError(cerror.ErrAPIGetPDClientFailed, errors.Trace(err))
 	}

cdc/server/server.go

@@ -151,7 +151,8 @@ func (s *server) prepare(ctx context.Context) error {
 				},
 				MinConnectTimeout: 3 * time.Second,
 			}),
-		))
+		),
+		pd.WithForwardingOption(config.EnablePDForwarding))
 	if err != nil {
 		return errors.Trace(err)
 	}

pkg/config/server_config.go

@@ -48,6 +48,11 @@ const (
 	// DisableMemoryLimit is the default max memory percentage for TiCDC server.
 	// 0 means no memory limit.
 	DisableMemoryLimit = 0
+
+	// EnablePDForwarding is the value of whether to enable PD client forwarding function.
+	// The PD client will forward the requests throughthe follower
+	// If there is a network partition problem between TiCDC and PD leader.
+	EnablePDForwarding = true
 )
 
 var (

pkg/migrate/migrate.go

@@ -147,7 +147,9 @@ func createPDClient(ctx context.Context,
 				},
 				MinConnectTimeout: 3 * time.Second,
 			}),
-		))
+		),
+		pd.WithForwardingOption(config.EnablePDForwarding),
+	)
 }
 
 // Note: we do not use etcd transaction to migrate key

pkg/upstream/upstream.go

@@ -26,6 +26,7 @@ import (
 	"github.com/pingcap/log"
 	tidbkv "github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tiflow/cdc/kv"
+	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/errorutil"
 	"github.com/pingcap/tiflow/pkg/etcd"
@@ -152,7 +153,8 @@ func initUpstream(ctx context.Context, up *Upstream, cfg CaptureTopologyCfg) err
 					},
 					MinConnectTimeout: 3 * time.Second,
 				}),
-			))
+			),
+			pd.WithForwardingOption(config.EnablePDForwarding))
 		if err != nil {
 			up.err.Store(err)
 			return errors.Trace(err)