sink(cdc): ddl sink errors shouldn't fail changefeed quickly (#9581)

cdc/entry/mounter.go

@@ -205,7 +205,7 @@ func (m *mounter) unmarshalAndMountRowChanged(ctx context.Context, raw *model.Ra
 		}
 		return nil, nil
 	}()
-	if err != nil && !cerror.IsChangefeedUnRetryableError(err) {
+	if err != nil && !cerror.ShouldFailChangefeed(err) {
 		log.Error("failed to mount and unmarshals entry, start to print debug info", zap.Error(err))
 		snap.PrintStatus(log.Error)
 	}

cdc/model/errors.go

@@ -28,7 +28,7 @@ type RunningError struct {
 	Message string    `json:"message"`
 }
 
-// IsChangefeedUnRetryableError return true if a running error contains a changefeed not retry error.
-func (r RunningError) IsChangefeedUnRetryableError() bool {
-	return cerror.IsChangefeedUnRetryableError(errors.New(r.Message + r.Code))
+// ShouldFailChangefeed return true if a running error contains a changefeed not retry error.
+func (r RunningError) ShouldFailChangefeed() bool {
+	return cerror.ShouldFailChangefeed(errors.New(r.Message + r.Code))
 }

cdc/model/errors_test.go

@@ -52,6 +52,6 @@ func TestIsChangefeedNotRetryError(t *testing.T) {
 	}
 
 	for _, c := range cases {
-		require.Equal(t, c.result, c.err.IsChangefeedUnRetryableError())
+		require.Equal(t, c.result, c.err.ShouldFailChangefeed())
 	}
 }

cdc/owner/ddl_sink.go

@@ -168,7 +168,7 @@ func (s *ddlSinkImpl) retrySinkAction(ctx context.Context, name string, action f
 		if err = action(); err == nil {
 			return nil
 		}
-		isRetryable := !cerror.IsChangefeedUnRetryableError(err) && errors.Cause(err) != context.Canceled
+		isRetryable := !cerror.ShouldFailChangefeed(err) && errors.Cause(err) != context.Canceled
 		log.Warn("owner ddl sink fails on action",
 			zap.String("namespace", s.changefeedID.Namespace),
 			zap.String("changefeed", s.changefeedID.ID),
@@ -400,7 +400,7 @@ func (s *ddlSinkImpl) emitSyncPoint(ctx context.Context, checkpointTs uint64) (e
 		if err == nil {
 			return nil
 		}
-		if !cerror.IsChangefeedUnRetryableError(err) && errors.Cause(err) != context.Canceled {
+		if !cerror.ShouldFailChangefeed(err) && errors.Cause(err) != context.Canceled {
 			// TODO(qupeng): retry it internally after async sink syncPoint is ready.
 			s.reportError(err)
 			return err

cdc/owner/feed_state_manager.go

@@ -483,7 +483,12 @@
 	// if there are a fastFail error in errs, we can just fastFail the changefeed
 	// and no need to patch other error to the changefeed info
 	for _, err := range errs {
+<<<<<<< HEAD
 		if cerrors.IsChangefeedFastFailErrorCode(errors.RFCErrorCode(err.Code)) {
+=======
+		if cerrors.IsChangefeedGCFastFailErrorCode(errors.RFCErrorCode(err.Code)) ||
+			err.ShouldFailChangefeed() {
+>>>>>>> dcfcb43a99 (sink(cdc): ddl sink errors shouldn't fail changefeed quickly (#9581))
 			m.state.PatchInfo(func(info *model.ChangeFeedInfo) (*model.ChangeFeedInfo, bool, error) {
 				if info == nil {
 					return nil, false, nil
@@ -495,7 +500,7 @@
 			m.patchState(model.StateFailed)
 			return
 		}
 	}
 
 	//  changefeed state from stopped to failed is allowed
 	// but stopped to error or normal is not allowed
@@ -512,20 +517,20 @@
 	// error in errs
 	for _, err := range errs {
 		if err.IsChangefeedUnRetryableError() {
 			m.state.PatchInfo(func(info *model.ChangeFeedInfo) (*model.ChangeFeedInfo, bool, error) {
 				if info == nil {
 					return nil, false, nil
 				}
 				info.Error = err
 				return info, true, nil
 			})
 			m.shouldBeRunning = false
 			m.patchState(model.StateError)
 			return
 		}
 	}
 
 	m.state.PatchInfo(func(info *model.ChangeFeedInfo) (*model.ChangeFeedInfo, bool, error) {
 		if info == nil {
 			return nil, false, nil
 		}
@@ -533,7 +538,7 @@
 			info.Error = err
 		}
 		return info, len(errs) > 0, nil
 	})
 
 	// If we enter into an abnormal state ('error', 'failed') for this changefeed now
 	// but haven't seen abnormal states in a sliding window (512 ticks),

cdc/processor/sinkmanager/manager.go

@@ -284,7 +284,7 @@ func (m *SinkManager) Run(ctx context.Context, warnings ...chan<- error) (err er
 		}
 
 		// If the error is retryable, we should retry to re-establish the internal resources.
-		if !cerror.IsChangefeedUnRetryableError(err) && errors.Cause(err) != context.Canceled {
+		if !cerror.ShouldFailChangefeed(err) && errors.Cause(err) != context.Canceled {
 			select {
 			case <-m.managerCtx.Done():
 			case warnings[0] <- err:

cdc/sink/ddlsink/mysql/mysql_ddl_sink.go

@@ -19,7 +19,6 @@ import (
 	"net/url"
 	"time"
 
-	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
 	timodel "github.com/pingcap/tidb/parser/model"
@@ -101,12 +100,7 @@ func NewDDLSink(
 
 // WriteDDLEvent writes a DDL event to the mysql database.
 func (m *DDLSink) WriteDDLEvent(ctx context.Context, ddl *model.DDLEvent) error {
-	err := m.execDDLWithMaxRetries(ctx, ddl)
-	// we should not retry changefeed if DDL failed by return an unretryable error.
-	if !errorutil.IsRetryableDDLError(err) {
-		return cerror.WrapChangefeedUnretryableErr(err)
-	}
-	return errors.Trace(err)
+	return m.execDDLWithMaxRetries(ctx, ddl)
 }
 
 func (m *DDLSink) execDDLWithMaxRetries(ctx context.Context, ddl *model.DDLEvent) error {

pkg/errors/helper.go

@@ -84,8 +84,8 @@ var changefeedUnRetryableErrors = []*errors.Error{
 	ErrStorageSinkInvalidConfig,
 }
 
-// IsChangefeedUnRetryableError returns true if an error is a changefeed not retry error.
-func IsChangefeedUnRetryableError(err error) bool {
+// ShouldFailChangefeed returns true if an error is a changefeed not retry error.
+func ShouldFailChangefeed(err error) bool {
 	for _, e := range changefeedUnRetryableErrors {
 		if e.Equal(err) {
 			return true

pkg/errors/helper_test.go

@@ -114,7 +114,7 @@ func TestChangefeedFastFailError(t *testing.T) {
 	require.Equal(t, false, IsChangefeedFastFailErrorCode(rfcCode))
 }
 
-func TestIsChangefeedUnRetryableError(t *testing.T) {
+func TestShouldFailChangefeed(t *testing.T) {
 	t.Parallel()
 	cases := []struct {
 		err      error
@@ -171,14 +171,14 @@ func TestIsChangefeedUnRetryableError(t *testing.T) {
 	}
 
 	for _, c := range cases {
-		require.Equal(t, c.expected, IsChangefeedUnRetryableError(c.err))
+		require.Equal(t, c.expected, ShouldFailChangefeed(c.err))
 	}
 
 	var code errors.RFCErrorCode
 	var ok bool
 	code, ok = RFCCode(ErrChangefeedUnretryable)
 	require.True(t, ok)
-	require.True(t, IsChangefeedUnRetryableError(errors.New(string(code))))
+	require.True(t, ShouldFailChangefeed(errors.New(string(code))))
 }
 
 func TestIsCliUnprintableError(t *testing.T) {

pkg/filter/expr_filter.go

@@ -415,7 +415,7 @@ func (f *dmlExprFilter) shouldSkipDML(
 	for _, rule := range rules {
 		ignore, err := rule.shouldSkipDML(row, rawRow, ti)
 		if err != nil {
-			if cerror.IsChangefeedUnRetryableError(err) {
+			if cerror.ShouldFailChangefeed(err) {
 				return false, err
 			}
 			return false, cerror.WrapError(cerror.ErrFailedToFilterDML, err, row)