owner(ticdc): Add backoff mechanism into changefeed restart logic

cdc/model/changefeed.go

@@ -94,16 +94,14 @@ func (s FeedState) IsNeeded(need string) bool {
 }
 
 const (
-	// errorHistoryGCInterval represents how long we keep error record in changefeed info
-	errorHistoryGCInterval = time.Minute * 10
-
-	// errorHistoryCheckInterval represents time window for failure check
-	errorHistoryCheckInterval = time.Minute * 2
-
 	// ErrorHistoryThreshold represents failure upper limit in time window.
 	// Before a changefeed is initialized, check the the failure count of this
 	// changefeed, if it is less than ErrorHistoryThreshold, then initialize it.
-	ErrorHistoryThreshold = 3
+	ErrorHistoryThreshold = 1
+
+	// deltaBackoffInterval is added to the time window to safely ensure that
+	// no errors occurred in that window
+	deltaBackoffInterval = 30 * time.Second
 )
 
 // ChangeFeedInfo describes the detail of a ChangeFeed
@@ -266,6 +264,7 @@ func (info *ChangeFeedInfo) VerifyAndComplete() error {
 	if info.Config.Consistent == nil {
 		info.Config.Consistent = defaultConfig.Consistent
 	}
+
 	return nil
 }
 
@@ -374,28 +373,6 @@ func (info *ChangeFeedInfo) fixSinkProtocol() {
 	}
 }
 
-// CheckErrorHistory checks error history of a changefeed
-// if having error record older than GC interval, set needSave to true.
-// if error counts reach threshold, set canInit to false.
-func (info *ChangeFeedInfo) CheckErrorHistory() (needSave bool, canInit bool) {
-	i := sort.Search(len(info.ErrorHis), func(i int) bool {
-		ts := info.ErrorHis[i]
-		return time.Since(time.Unix(ts/1e3, (ts%1e3)*1e6)) < errorHistoryGCInterval
-	})
-	info.ErrorHis = info.ErrorHis[i:]
-
-	if i > 0 {
-		needSave = true
-	}
-
-	i = sort.Search(len(info.ErrorHis), func(i int) bool {
-		ts := info.ErrorHis[i]
-		return time.Since(time.Unix(ts/1e3, (ts%1e3)*1e6)) < errorHistoryCheckInterval
-	})
-	canInit = len(info.ErrorHis)-i < ErrorHistoryThreshold
-	return
-}
-
 // HasFastFailError returns true if the error in changefeed is fast-fail
 func (info *ChangeFeedInfo) HasFastFailError() bool {
 	if info.Error == nil {
@@ -404,26 +381,26 @@ func (info *ChangeFeedInfo) HasFastFailError() bool {
 	return cerror.ChangefeedFastFailErrorCode(errors.RFCErrorCode(info.Error.Code))
 }
 
-// findActiveErrors finds all errors occurring within errorHistoryCheckInterval
-func (info *ChangeFeedInfo) findActiveErrors() []int64 {
+// findActiveErrors finds all errors occurring within backoffInterval
+func (info *ChangeFeedInfo) findActiveErrors(backoffInterval time.Duration) []int64 {
 	i := sort.Search(len(info.ErrorHis), func(i int) bool {
 		ts := info.ErrorHis[i]
-		// ts is a errors occurrence time, here to find all errors occurring within errorHistoryCheckInterval
-		return time.Since(time.Unix(ts/1e3, (ts%1e3)*1e6)) < errorHistoryCheckInterval
+		// ts is a errors occurrence time, here to find all errors occurring within a time window
+		return time.Since(time.Unix(ts/1e3, (ts%1e3)*1e6)) < backoffInterval+deltaBackoffInterval
 	})
 	return info.ErrorHis[i:]
 }
 
 // ErrorsReachedThreshold checks error history of a changefeed
 // returns true if error counts reach threshold
-func (info *ChangeFeedInfo) ErrorsReachedThreshold() bool {
-	return len(info.findActiveErrors()) >= ErrorHistoryThreshold
+func (info *ChangeFeedInfo) ErrorsReachedThreshold(backoffInterval time.Duration) bool {
+	return len(info.findActiveErrors(backoffInterval)) >= ErrorHistoryThreshold
 }
 
 // CleanUpOutdatedErrorHistory cleans up the outdated error history
 // return true if the ErrorHis changed
-func (info *ChangeFeedInfo) CleanUpOutdatedErrorHistory() bool {
+func (info *ChangeFeedInfo) CleanUpOutdatedErrorHistory(backoffInterval time.Duration) bool {
 	lastLenOfErrorHis := len(info.ErrorHis)
-	info.ErrorHis = info.findActiveErrors()
+	info.ErrorHis = info.findActiveErrors(backoffInterval)
 	return lastLenOfErrorHis != len(info.ErrorHis)
 }

cdc/model/changefeed_test.go

@@ -641,34 +641,6 @@ func TestChangeFeedInfoClone(t *testing.T) {
 	require.True(t, info.Config.EnableOldValue)
 }
 
-func TestCheckErrorHistory(t *testing.T) {
-	t.Parallel()
-
-	now := time.Now()
-	info := &ChangeFeedInfo{
-		ErrorHis: []int64{},
-	}
-	for i := 0; i < 5; i++ {
-		tm := now.Add(-errorHistoryGCInterval)
-		info.ErrorHis = append(info.ErrorHis, tm.UnixNano()/1e6)
-		time.Sleep(time.Millisecond)
-	}
-	for i := 0; i < ErrorHistoryThreshold-1; i++ {
-		info.ErrorHis = append(info.ErrorHis, time.Now().UnixNano()/1e6)
-		time.Sleep(time.Millisecond)
-	}
-	time.Sleep(time.Millisecond)
-	needSave, canInit := info.CheckErrorHistory()
-	require.True(t, needSave)
-	require.True(t, canInit)
-	require.Equal(t, ErrorHistoryThreshold-1, len(info.ErrorHis))
-
-	info.ErrorHis = append(info.ErrorHis, time.Now().UnixNano()/1e6)
-	needSave, canInit = info.CheckErrorHistory()
-	require.False(t, needSave)
-	require.False(t, canInit)
-}
-
 func TestChangefeedInfoStringer(t *testing.T) {
 	t.Parallel()
 

cdc/owner/changefeed.go

@@ -84,7 +84,7 @@ func newChangefeed(id model.ChangeFeedID, gcManager gc.Manager) *changefeed {
 		// The scheduler will be created lazily.
 		scheduler:        nil,
 		barriers:         newBarriers(),
-		feedStateManager: new(feedStateManager),
+		feedStateManager: newFeedStateManager(),
 		gcManager:        gcManager,
 
 		errCh:  make(chan error, defaultErrChSize),

cdc/owner/feed_state_manager.go

@@ -16,6 +16,7 @@ package owner
 import (
 	"time"
 
+	"github.com/cenkalti/backoff/v4"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
@@ -24,6 +25,18 @@ import (
 	"go.uber.org/zap"
 )
 
+const (
+	// When errors occurred and we need to do backoff, we start an exponential backoff
+	// with an interval from 10s to 30min (10s, 20s, 40s, ... 30min, 30min...).
+	// And when the duration (now - backoff start) exceeds 1 hour,
+	// the backoff will be stopped. To avoid thunderherd, a random factor is added to backoff interval.
+	defaultBackoffInitInterval        = 10 * time.Second
+	defaultBackoffMaxInterval         = 30 * time.Minute
+	defaultBackoffMaxElapsedTime      = 1 * time.Hour
+	defaultBackoffRandomizationFactor = 0.1
+	defaultBackoffMultiplier          = 2.0
+)
+
 // feedStateManager manages the ReactorState of a changefeed
 // when an error or an admin job occurs, the feedStateManager is responsible for controlling the ReactorState
 type feedStateManager struct {
@@ -34,7 +47,33 @@ type feedStateManager struct {
 	// shouldBeRemoved = false means the changefeed is paused
 	shouldBeRemoved bool
 
-	adminJobQueue []*model.AdminJob
+	adminJobQueue   []*model.AdminJob
+	lastErrorTime   time.Time
+	backoffInterval time.Duration
+	errBackoff      *backoff.ExponentialBackOff
+}
+
+// newFeedStateManager creates feedStateManager and initialize the exponential backoff
+func newFeedStateManager() *feedStateManager {
+	f := new(feedStateManager)
+
+	f.errBackoff = backoff.NewExponentialBackOff()
+	f.errBackoff.InitialInterval = defaultBackoffInitInterval
+	f.errBackoff.MaxInterval = defaultBackoffMaxInterval
+	f.errBackoff.MaxElapsedTime = defaultBackoffMaxElapsedTime
+	f.errBackoff.Multiplier = defaultBackoffMultiplier
+	f.errBackoff.RandomizationFactor = defaultBackoffRandomizationFactor
+
+	f.resetErrBackoff()
+	f.lastErrorTime = time.Unix(0, 0)
+
+	return f
+}
+
+func (m *feedStateManager) resetErrBackoff() {
+	m.errBackoff.Reset()
+
+	m.backoffInterval = m.errBackoff.NextBackOff()
 }
 
 func (m *feedStateManager) Tick(state *orchestrator.ChangefeedReactorState) {
@@ -145,9 +184,14 @@ func (m *feedStateManager) handleAdminJob() (jobsPending bool) {
 			return
 		}
 		m.shouldBeRunning = true
+		// when the changefeed is manually resumed, we must reset the backoff
+		m.resetErrBackoff()
+		// The lastErrorTime also needs to be cleared before a fresh run.
+		m.lastErrorTime = time.Unix(0, 0)
 		jobsPending = true
 		m.patchState(model.StateNormal)
-		// remove error history to make sure the changefeed can running in next tick
+		// to make sure the changefeed can running in next tick,
+		// we remove the error history
 		m.state.PatchInfo(func(info *model.ChangeFeedInfo) (*model.ChangeFeedInfo, bool, error) {
 			if info == nil {
 				return nil, false, nil
@@ -287,7 +331,7 @@ func (m *feedStateManager) handleError(errs ...*model.RunningError) {
 				}
 				info.Error = err
 				info.ErrorHis = append(info.ErrorHis, time.Now().UnixNano()/1e6)
-				info.CleanUpOutdatedErrorHistory()
+				info.CleanUpOutdatedErrorHistory(m.backoffInterval)
 				return info, true, nil
 			})
 			m.shouldBeRunning = false
@@ -304,14 +348,38 @@ func (m *feedStateManager) handleError(errs ...*model.RunningError) {
 			info.Error = err
 			info.ErrorHis = append(info.ErrorHis, time.Now().UnixNano()/1e6)
 		}
-		changed := info.CleanUpOutdatedErrorHistory()
+		changed := info.CleanUpOutdatedErrorHistory(m.backoffInterval)
 		return info, changed || len(errs) > 0, nil
 	})
 
-	// if the number of errors has reached the error threshold, stop the changefeed
-	if m.state.Info.ErrorsReachedThreshold() {
+	if len(errs) > 0 {
+		m.lastErrorTime = time.Now()
+		if !m.state.Info.ErrorsReachedThreshold(m.backoffInterval) {
+			m.resetErrBackoff()
+		}
+	}
+
+	if m.lastErrorTime == time.Unix(0, 0) {
+		return
+	}
+
+	if time.Since(m.lastErrorTime) < m.backoffInterval {
 		m.shouldBeRunning = false
 		m.patchState(model.StateError)
-		return
+	} else {
+		m.backoffInterval = m.errBackoff.NextBackOff()
+		m.lastErrorTime = time.Unix(0, 0)
+
+		// if the duration since backoff start exceeds MaxElapsedTime,
+		// we set the state of changefeed to "failed" and don't let it run again unless it is manually resumed.
+		if m.backoffInterval == backoff.Stop {
+			log.Warn("changefeed will not be restarted because it has been failing for a long time period",
+				zap.Duration("maxElapsedTime", m.errBackoff.MaxElapsedTime))
+			m.shouldBeRunning = false
+			m.patchState(model.StateFailed)
+		} else {
+			log.Info("changefeed restart backoff interval is changed", zap.String("changefeedID", m.state.ID),
+				zap.Duration("backoffInterval", m.backoffInterval))
+		}
 	}
 }

cdc/owner/feed_state_manager_test.go

@@ -29,7 +29,7 @@ type feedStateManagerSuite struct{}
 func (s *feedStateManagerSuite) TestHandleJob(c *check.C) {
 	defer testleak.AfterTest(c)()
 	ctx := cdcContext.NewBackendContext4Test(true)
-	manager := new(feedStateManager)
+	manager := newFeedStateManager()
 	state := orchestrator.NewChangefeedReactorState(ctx.ChangefeedVars().ID)
 	tester := orchestrator.NewReactorStateTester(c, state, nil)
 	state.PatchInfo(func(info *model.ChangeFeedInfo) (*model.ChangeFeedInfo, bool, error) {
@@ -104,7 +104,7 @@ func (s *feedStateManagerSuite) TestHandleJob(c *check.C) {
 func (s *feedStateManagerSuite) TestMarkFinished(c *check.C) {
 	defer testleak.AfterTest(c)()
 	ctx := cdcContext.NewBackendContext4Test(true)
-	manager := new(feedStateManager)
+	manager := newFeedStateManager()
 	state := orchestrator.NewChangefeedReactorState(ctx.ChangefeedVars().ID)
 	tester := orchestrator.NewReactorStateTester(c, state, nil)
 	state.PatchInfo(func(info *model.ChangeFeedInfo) (*model.ChangeFeedInfo, bool, error) {
@@ -132,7 +132,7 @@ func (s *feedStateManagerSuite) TestMarkFinished(c *check.C) {
 func (s *feedStateManagerSuite) TestCleanUpInfos(c *check.C) {
 	defer testleak.AfterTest(c)()
 	ctx := cdcContext.NewBackendContext4Test(true)
-	manager := new(feedStateManager)
+	manager := newFeedStateManager()
 	state := orchestrator.NewChangefeedReactorState(ctx.ChangefeedVars().ID)
 	tester := orchestrator.NewReactorStateTester(c, state, nil)
 	state.PatchInfo(func(info *model.ChangeFeedInfo) (*model.ChangeFeedInfo, bool, error) {
@@ -175,7 +175,7 @@ func (s *feedStateManagerSuite) TestCleanUpInfos(c *check.C) {
 func (s *feedStateManagerSuite) TestHandleError(c *check.C) {
 	defer testleak.AfterTest(c)()
 	ctx := cdcContext.NewBackendContext4Test(true)
-	manager := new(feedStateManager)
+	manager := newFeedStateManager()
 	state := orchestrator.NewChangefeedReactorState(ctx.ChangefeedVars().ID)
 	tester := orchestrator.NewReactorStateTester(c, state, nil)
 	state.PatchInfo(func(info *model.ChangeFeedInfo) (*model.ChangeFeedInfo, bool, error) {
@@ -229,7 +229,7 @@ func (s *feedStateManagerSuite) TestHandleError(c *check.C) {
 func (s *feedStateManagerSuite) TestChangefeedStatusNotExist(c *check.C) {
 	defer testleak.AfterTest(c)()
 	ctx := cdcContext.NewBackendContext4Test(true)
-	manager := new(feedStateManager)
+	manager := newFeedStateManager()
 	state := orchestrator.NewChangefeedReactorState(ctx.ChangefeedVars().ID)
 	tester := orchestrator.NewReactorStateTester(c, state, map[string]string{
 		"/tidb/cdc/capture/d563bfc0-f406-4f34-bc7d-6dc2e35a44e5": `{"id":"d563bfc0-f406-4f34-bc7d-6dc2e35a44e5","address":"172.16.6.147:8300","version":"v5.0.0-master-dirty"}`,

cdc/sink/codec/schema_registry.go

@@ -25,7 +25,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/cenkalti/backoff"
+	"github.com/cenkalti/backoff/v4"
 	"github.com/linkedin/goavro/v2"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/aws/aws-sdk-go v1.35.3
 	github.com/benbjohnson/clock v1.1.0
 	github.com/bradleyjkemp/grpc-tools v0.2.5
-	github.com/cenkalti/backoff v2.2.1+incompatible
+	github.com/cenkalti/backoff/v4 v4.0.2
 	github.com/chaos-mesh/go-sqlsmith v0.0.0-20211025024535-03ae33408684
 	github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e
 	github.com/cockroachdb/pebble v0.0.0-20210719141320-8c3bd06debb5

go.sum

@@ -137,8 +137,7 @@ github.com/cakturk/go-netstat v0.0.0-20200220111822-e5b49efee7a5 h1:BjkPE3785EwP
 github.com/cakturk/go-netstat v0.0.0-20200220111822-e5b49efee7a5/go.mod h1:jtAfVaU/2cu1+wdSRPWE2c1N2qeAA3K4RH9pYgqwets=
 github.com/carlmjohnson/flagext v0.21.0 h1:/c4uK3ie786Z7caXLcIMvePNSSiH3bQVGDvmGLMme60=
 github.com/carlmjohnson/flagext v0.21.0/go.mod h1:Eenv0epIUAr4NuedNmkzI8WmBmjIxZC239XcKxYS2ac=
-github.com/cenkalti/backoff v2.2.1+incompatible h1:tNowT99t7UNflLxfYYSlKYsBpXdEet03Pg2g16Swow4=
-github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
+github.com/cenkalti/backoff/v4 v4.0.2 h1:JIufpQLbh4DkbQoii76ItQIUFzevQSqOLZca4eamEDs=
 github.com/cenkalti/backoff/v4 v4.0.2/go.mod h1:eEew/i+1Q6OrCDZh3WiXYv3+nJwBASZ8Bog/87DQnVg=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=

pkg/config/cyclic.go

@@ -33,7 +33,7 @@ func (c *CyclicConfig) IsEnabled() bool {
 	return c != nil && c.Enable
 }
 
-// Marshal returns the json marshal format of a ReplicationConfig
+// Marshal returns the json marshal format of a CyclicConfig
 func (c *CyclicConfig) Marshal() (string, error) {
 	cfg, err := json.Marshal(c)
 	if err != nil {
@@ -42,7 +42,7 @@ func (c *CyclicConfig) Marshal() (string, error) {
 	return string(cfg), nil
 }
 
-// Unmarshal unmarshals into *ReplicationConfig from json marshal byte slice
+// Unmarshal unmarshals into *CyclicConfig from json marshal byte slice
 func (c *CyclicConfig) Unmarshal(data []byte) error {
 	return json.Unmarshal(data, c)
 }

tests/mq_protocol_tests/framework/dsl.go

@@ -17,7 +17,7 @@ import (
 	"context"
 	"time"
 
-	backoff2 "github.com/cenkalti/backoff"
+	"github.com/cenkalti/backoff/v4"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"go.uber.org/zap"
@@ -90,8 +90,8 @@ func (b *basicAwaitable) Wait() Checkable {
 	}
 	defer cancel()
 
-	backoff := backoff2.NewExponentialBackOff()
-	backoff.MaxInterval = waitMaxPollInterval
+	expBackoff := backoff.NewExponentialBackOff()
+	expBackoff.MaxInterval = waitMaxPollInterval
 	for {
 		select {
 		case <-ctx.Done():
@@ -109,8 +109,8 @@ func (b *basicAwaitable) Wait() Checkable {
 			return b
 		}
 
-		interval := backoff.NextBackOff()
-		if interval == backoff2.Stop {
+		interval := expBackoff.NextBackOff()
+		if interval == backoff.Stop {
 			return &errorCheckableAndAwaitable{errors.New("Maximum retry interval reached")}
 		}
 		log.Debug("Wait(): pollable returned false, backing off", zap.Duration("interval", interval))