sink(cdc): clean backends if table sink is stuck too long (#9527) (#9548

) close #9542
pingcap · Aug 15, 2023 · 8622904 · 8622904
1 parent db5fb4d
commit 8622904
Show file tree

Hide file tree

Showing 25 changed files with 406 additions and 192 deletions.
diff --git a/cdc/api/v2/model.go b/cdc/api/v2/model.go
@@ -25,6 +25,7 @@ import (
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/integrity"
 	"github.com/pingcap/tiflow/pkg/security"
+	"github.com/pingcap/tiflow/pkg/util"
 )
 
 // EmptyResponse return empty {} to http client
@@ -369,6 +370,9 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 			CloudStorageConfig:       cloudStorageConfig,
 			SafeMode:                 c.Sink.SafeMode,
 		}
+		if c.Sink.AdvanceTimeoutInSec != nil {
+			res.Sink.AdvanceTimeoutInSec = util.AddressOf(*c.Sink.AdvanceTimeoutInSec)
+		}
 	}
 	if c.Mounter != nil {
 		res.Mounter = &config.MounterConfig{
@@ -572,6 +576,9 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 			CloudStorageConfig:       cloudStorageConfig,
 			SafeMode:                 cloned.Sink.SafeMode,
 		}
+		if cloned.Sink.AdvanceTimeoutInSec != nil {
+			res.Sink.AdvanceTimeoutInSec = util.AddressOf(*cloned.Sink.AdvanceTimeoutInSec)
+		}
 	}
 	if cloned.Consistent != nil {
 		res.Consistent = &ConsistentConfig{
@@ -722,6 +729,7 @@ type SinkConfig struct {
 	KafkaConfig              *KafkaConfig        `json:"kafka_config,omitempty"`
 	MySQLConfig              *MySQLConfig        `json:"mysql_config,omitempty"`
 	CloudStorageConfig       *CloudStorageConfig `json:"cloud_storage_config,omitempty"`
+	AdvanceTimeoutInSec      *uint               `json:"advance_timeout,omitempty"`
 }
 
 // CSVConfig denotes the csv config

diff --git a/cdc/api/v2/model_test.go b/cdc/api/v2/model_test.go
@@ -23,6 +23,7 @@ import (
 	filter "github.com/pingcap/tidb/util/table-filter"
 	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/redo"
+	"github.com/pingcap/tiflow/pkg/util"
 	"github.com/stretchr/testify/require"
 )
 
@@ -53,6 +54,7 @@ var defaultAPIConfig = &ReplicaConfig{
 		DateSeparator:            config.DateSeparatorDay.String(),
 		EnablePartitionSeparator: true,
 		EnableKafkaSinkV2:        false,
+		AdvanceTimeoutInSec:      util.AddressOf(uint(150)),
 	},
 	Consistent: &ConsistentConfig{
 		Level:             "none",

diff --git a/cdc/processor/processor_test.go b/cdc/processor/processor_test.go
@@ -133,7 +133,8 @@ func initProcessor4Test(
         },
         "sink": {
             "dispatchers": null,
-            "protocol": "open-protocol"
+            "protocol": "open-protocol",
+            "advance-timeout-in-sec": 150
         }
     },
     "state": "normal",