config (ticdc): add checking for changefeed scheduler config (#9646)

close #9645

cdc/puller/ddl_puller.go

@@ -396,6 +396,16 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 				zap.String("query", job.Query),
 				zap.Stringer("job", job))
 		}
+		if err != nil {
+			log.Warn("handle ddl job failed",
+				zap.String("namespace", p.changefeedID.Namespace),
+				zap.String("changefeed", p.changefeedID.ID),
+				zap.String("schema", job.SchemaName),
+				zap.String("table", job.TableName),
+				zap.String("query", job.Query),
+				zap.Stringer("job", job),
+				zap.Error(err))
+		}
 	}()
 
 	if job.BinlogInfo.FinishedTS <= p.getResolvedTs() ||
@@ -424,6 +434,7 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 		if discard {
 			return true, nil
 		}
+		return true, errors.Trace(err)
 	}
 
 	switch job.Type {

metrics/alertmanager/ticdc.rules.yml

@@ -110,4 +110,16 @@ groups:
     annotations:
       description: 'cluster: ENV_LABELS_ENV, instance: {{ $labels.instance }}, values: {{ $value }}'
       value: '{{ $value }}'
-      summary: tikv cdc scan duration seconds more than 10 min
+      summary: tikv cdc scan duration seconds more than 10 min
+
+  - alert: ticdc_memory_abnormal
+    expr: go_memstats_heap_alloc_bytes{job="ticdc"} > 1e+10
+    for: 1m
+    labels:
+      env: ENV_LABELS_ENV
+      level: warning
+      expr: go_memstats_heap_alloc_bytes{job="ticdc"} > 1e+10
+    annotations:
+      description: 'cluster: ENV_LABELS_ENV, instance: {{ $labels.instance }}, values:{{ $value }}'
+      value: '{{ $value }}'
+      summary: TiCDC heap memory usage is over 10 GB

pkg/config/replica_config.go

@@ -239,6 +239,11 @@ func (c *ReplicaConfig) ValidateAndAdjust(sinkURI *url.URL) error { // check sin
 	}
 	if c.Scheduler == nil {
 		c.FixScheduler(false)
+	} else {
+		err := c.Scheduler.Validate()
+		if err != nil {
+			return err
+		}
 	}
 	// TODO: Remove the hack once span replication is compatible with all sinks.
 	if !isSinkCompatibleWithSpanReplication(sinkURI) {

pkg/config/replica_config_test.go

@@ -235,6 +235,13 @@ func TestReplicaConfigValidate(t *testing.T) {
 	err = conf.ValidateAndAdjust(sinkURL)
 	require.NoError(t, err)
 	require.Equal(t, uint64(1024), conf.MemoryQuota)
+
+	conf.Scheduler = &ChangefeedSchedulerConfig{
+		EnableTableAcrossNodes: true,
+		RegionThreshold:        -1,
+	}
+	err = conf.ValidateAndAdjust(sinkURL)
+	require.Error(t, err)
 }
 
 func TestValidateAndAdjust(t *testing.T) {

pkg/config/scheduler_config.go

@@ -14,6 +14,7 @@
 package config
 
 import (
+	"errors"
 	"time"
 
 	cerror "github.com/pingcap/tiflow/pkg/errors"
@@ -32,6 +33,20 @@ type ChangefeedSchedulerConfig struct {
 	RegionPerSpan int `toml:"region-per-span" json:"region-per-span"`
 }
 
+// Validate validates the config.
+func (c *ChangefeedSchedulerConfig) Validate() error {
+	if !c.EnableTableAcrossNodes {
+		return nil
+	}
+	if c.RegionThreshold < 0 {
+		return errors.New("region-threshold must be larger than 0")
+	}
+	if c.WriteKeyThreshold < 0 {
+		return errors.New("write-key-threshold must be larger than 0")
+	}
+	return nil
+}
+
 // SchedulerConfig configs TiCDC scheduler.
 type SchedulerConfig struct {
 	// HeartbeatTick is the number of owner tick to initial a heartbeat to captures.
@@ -91,6 +106,5 @@ func (c *SchedulerConfig) ValidateAndAdjust() error {
 		return cerror.ErrInvalidServerOption.GenWithStackByArgs(
 			"add-table-batch-size must be large than 0")
 	}
-
 	return nil
 }