errors

Documentation

Index

• Variables
• func GRPCStatusCode(err error) codes.Code
• func HTTPStatusCode(err error) int
• func IsChangefeedGCFastFailError(err error) bool
• func IsChangefeedGCFastFailErrorCode(errCode errors.RFCErrorCode) bool
• func IsCliUnprintableError(err error) bool
• func IsContextCanceledError(err error) bool
• func IsContextDeadlineExceededError(err error) bool
• func IsDupEntryError(err error) bool
• func IsRetryableError(err error) bool
• func OriginError(err error) error
• func RFCCode(err error) (errors.RFCErrorCode, bool)
• func ShouldFailChangefeed(err error) bool
• func ToDMError(err error) error
• func WrapChangefeedUnretryableErr(err error, args ...interface{}) error
• func WrapError(rfcError *errors.Error, err error, args ...interface{}) error

Variables

var (
	// kv related errors
	ErrChangeFeedNotExists = errors.Normalize(
		"changefeed not exists, %s",
		errors.RFCCodeText("CDC:ErrChangeFeedNotExists"),
	)
	ErrChangeFeedAlreadyExists = errors.Normalize(
		"changefeed already exists, %s",
		errors.RFCCodeText("CDC:ErrChangeFeedAlreadyExists"),
	)
	ErrChangeFeedDeletionUnfinished = errors.Normalize(
		"changefeed exists after deletion, %s",
		errors.RFCCodeText("CDC:ErrChangeFeedDeletionUnfinished"),
	)
	ErrCaptureNotExist = errors.Normalize(
		"capture not exists, %s",
		errors.RFCCodeText("CDC:ErrCaptureNotExist"),
	)
	ErrSchedulerRequestFailed = errors.Normalize(
		"scheduler request failed, %s",
		errors.RFCCodeText("CDC:ErrSchedulerRequestFailed"),
	)
	ErrGetAllStoresFailed = errors.Normalize(
		"get stores from pd failed",
		errors.RFCCodeText("CDC:ErrGetAllStoresFailed"),
	)
	ErrMetaListDatabases = errors.Normalize(
		"meta store list databases",
		errors.RFCCodeText("CDC:ErrMetaListDatabases"),
	)
	ErrDDLSchemaNotFound = errors.Normalize(
		"cannot find mysql.tidb_ddl_job schema",
		errors.RFCCodeText("CDC:ErrDDLSchemaNotFound"),
	)
	ErrDDLUnsupportType = errors.Normalize(
		"unsupport ddl type %s, query %s",
		errors.RFCCodeText("CDC:ErrDDLUnsupportType"),
	)
	ErrGRPCDialFailed = errors.Normalize(
		"grpc dial failed",
		errors.RFCCodeText("CDC:ErrGRPCDialFailed"),
	)
	ErrTiKVEventFeed = errors.Normalize(
		"tikv event feed failed",
		errors.RFCCodeText("CDC:ErrTiKVEventFeed"),
	)
	ErrPDBatchLoadRegions = errors.Normalize(
		"pd batch load regions failed",
		errors.RFCCodeText("CDC:ErrPDBatchLoadRegions"),
	)
	ErrMetaNotInRegion = errors.Normalize(
		"meta not exists in region",
		errors.RFCCodeText("CDC:ErrMetaNotInRegion"),
	)
	ErrRegionsNotCoverSpan = errors.Normalize(
		"regions not completely left cover span, span %v regions: %v",
		errors.RFCCodeText("CDC:ErrRegionsNotCoverSpan"),
	)
	ErrGetTiKVRPCContext = errors.Normalize(
		"get tikv grpc context failed",
		errors.RFCCodeText("CDC:ErrGetTiKVRPCContext"),
	)
	ErrPendingRegionCancel = errors.Normalize(
		"pending region cancelled due to stream disconnecting",
		errors.RFCCodeText("CDC:ErrPendingRegionCancel"),
	)
	ErrEventFeedAborted = errors.Normalize(
		"single event feed aborted",
		errors.RFCCodeText("CDC:ErrEventFeedAborted"),
	)
	ErrUnknownKVEventType = errors.Normalize(
		"unknown kv optype: %s, entry: %v",
		errors.RFCCodeText("CDC:ErrUnknownKVEventType"),
	)
	ErrPrewriteNotMatch = errors.Normalize(
		"prewrite not match, key: %s, start-ts: %d, commit-ts: %d, type: %s, optype: %s",
		errors.RFCCodeText("CDC:ErrPrewriteNotMatch"),
	)
	ErrEventFeedEventError = errors.Normalize(
		"eventfeed returns event error",
		errors.RFCCodeText("CDC:ErrEventFeedEventError"),
	)
	ErrPDEtcdAPIError = errors.Normalize(
		"etcd api call error",
		errors.RFCCodeText("CDC:ErrPDEtcdAPIError"),
	)
	ErrNewStore = errors.Normalize(
		"new store failed",
		errors.RFCCodeText("CDC:ErrNewStore"),
	)
	ErrRegionWorkerExit = errors.Normalize(
		"region worker exited",
		errors.RFCCodeText("CDC:ErrRegionWorkerExit"),
	)

	// codec related errors
	ErrEncodeFailed = errors.Normalize(
		"encode failed",
		errors.RFCCodeText("CDC:ErrEncodeFailed"),
	)
	ErrDecodeFailed = errors.Normalize(
		"decode failed: %s",
		errors.RFCCodeText("CDC:ErrDecodeFailed"),
	)
	ErrFilterRuleInvalid = errors.Normalize(
		"filter rule is invalid %v",
		errors.RFCCodeText("CDC:ErrFilterRuleInvalid"),
	)

	ErrDispatcherFailed = errors.Normalize(
		"dispatcher failed",
		errors.RFCCodeText("CDC:ErrDispatcherFailed"),
	)

	ErrColumnSelectorFailed = errors.Normalize(
		"column selector failed",
		errors.RFCCodeText("CDC:ErrColumnSelectorFailed"),
	)

	// internal errors
	ErrAdminStopProcessor = errors.Normalize(
		"stop processor by admin command",
		errors.RFCCodeText("CDC:ErrAdminStopProcessor"),
	)
	ErrCodeNilFunction = errors.Normalize(
		"function is not initialized",
		errors.RFCCodeText("CDC:ErrCodeNilFunction"),
	)

	// Errors caused by unexpected behavior from external systems
	ErrTiDBUnexpectedJobMeta = errors.Normalize(
		"unexpected `job_meta` from tidb",
		errors.RFCCodeText("CDC:ErrTiDBUnexpectedJobMeta"),
	)

	// ErrVersionIncompatible is an error for running CDC on an incompatible Cluster.
	ErrVersionIncompatible = errors.Normalize(
		"version is incompatible: %s",
		errors.RFCCodeText("CDC:ErrVersionIncompatible"),
	)
	ErrClusterIDMismatch = errors.Normalize(
		"cluster ID mismatch, tikv cluster ID is %d and request cluster ID is %d",
		errors.RFCCodeText("CDC:ErrClusterIDMismatch"),
	)
	ErrMultipleCDCClustersExist = errors.Normalize(
		"multiple TiCDC clusters exist while using --pd",
		errors.RFCCodeText("CDC:ErrMultipleCDCClustersExist"),
	)

	// sink related errors
	ErrExecDDLFailed = errors.Normalize(
		"exec DDL failed",
		errors.RFCCodeText("CDC:ErrExecDDLFailed"),
	)
	ErrKafkaSendMessage = errors.Normalize(
		"kafka send message failed",
		errors.RFCCodeText("CDC:ErrKafkaSendMessage"),
	)
	ErrKafkaProducerClosed = errors.Normalize(
		"kafka producer closed",
		errors.RFCCodeText("CDC:ErrKafkaProducerClosed"),
	)
	ErrKafkaAsyncSendMessage = errors.Normalize(
		"kafka async send message failed",
		errors.RFCCodeText("CDC:ErrKafkaAsyncSendMessage"),
	)
	ErrKafkaInvalidPartitionNum = errors.Normalize(
		"invalid partition num %d",
		errors.RFCCodeText("CDC:ErrKafkaInvalidPartitionNum"),
	)
	ErrKafkaInvalidRequiredAcks = errors.Normalize(
		"invalid required acks %d, "+
			"only support these values: 0(NoResponse),1(WaitForLocal) and -1(WaitForAll)",
		errors.RFCCodeText("CDC:ErrKafkaInvalidRequiredAcks"),
	)
	ErrKafkaNewProducer = errors.Normalize(
		"new kafka producer",
		errors.RFCCodeText("CDC:ErrKafkaNewProducer"),
	)
	ErrKafkaInvalidClientID = errors.Normalize(
		"invalid kafka client ID '%s'",
		errors.RFCCodeText("CDC:ErrKafkaInvalidClientID"),
	)
	ErrKafkaInvalidVersion = errors.Normalize(
		"invalid kafka version",
		errors.RFCCodeText("CDC:ErrKafkaInvalidVersion"),
	)
	ErrKafkaInvalidConfig = errors.Normalize(
		"kafka config invalid",
		errors.RFCCodeText("CDC:ErrKafkaInvalidConfig"),
	)
	ErrKafkaCreateTopic = errors.Normalize(
		"kafka create topic failed",
		errors.RFCCodeText("CDC:ErrKafkaCreateTopic"),
	)
	ErrKafkaInvalidTopicExpression = errors.Normalize(
		"invalid topic expression: %s ",
		errors.RFCCodeText("CDC:ErrKafkaTopicExprInvalid"),
	)
	ErrKafkaConfigNotFound = errors.Normalize(
		"kafka config item not found",
		errors.RFCCodeText("CDC:ErrKafkaConfigNotFound"),
	)
	// for pulsar
	ErrPulsarSendMessage = errors.Normalize(
		"pulsar send message failed",
		errors.RFCCodeText("CDC:ErrPulsarSendMessage"),
	)
	ErrPulsarProducerClosed = errors.Normalize(
		"pulsar producer closed",
		errors.RFCCodeText("CDC:ErrPulsarProducerClosed"),
	)
	ErrPulsarAsyncSendMessage = errors.Normalize(
		"pulsar async send message failed",
		errors.RFCCodeText("CDC:ErrPulsarAsyncSendMessage"),
	)
	ErrPulsarFlushUnfinished = errors.Normalize(
		"flush not finished before producer close",
		errors.RFCCodeText("CDC:ErrPulsarFlushUnfinished"),
	)
	ErrPulsarInvalidPartitionNum = errors.Normalize(
		"invalid partition num %d",
		errors.RFCCodeText("CDC:ErrPulsarInvalidPartitionNum"),
	)
	ErrPulsarNewClient = errors.Normalize(
		"new pulsar client",
		errors.RFCCodeText("CDC:ErrPulsarNewClient"),
	)
	ErrPulsarNewProducer = errors.Normalize(
		"new pulsar producer",
		errors.RFCCodeText("CDC:ErrPulsarNewProducer"),
	)
	ErrPulsarInvalidClientID = errors.Normalize(
		"invalid pulsar client ID '%s'",
		errors.RFCCodeText("CDC:ErrPulsarInvalidClientID"),
	)
	ErrPulsarInvalidVersion = errors.Normalize(
		"invalid pulsar version",
		errors.RFCCodeText("CDC:ErrPulsarInvalidVersion"),
	)
	ErrPulsarInvalidConfig = errors.Normalize(
		"pulsar config invalid %s",
		errors.RFCCodeText("CDC:ErrPulsarInvalidConfig"),
	)
	ErrPulsarCreateTopic = errors.Normalize(
		"pulsar create topic failed",
		errors.RFCCodeText("CDC:ErrPulsarCreateTopic"),
	)
	ErrPulsarInvalidTopicExpression = errors.Normalize(
		"invalid topic expression",
		errors.RFCCodeText("CDC:ErrPulsarTopicExprInvalid"),
	)
	ErrPulsarBrokerConfigNotFound = errors.Normalize(
		"pulsar broker config item not found",
		errors.RFCCodeText("CDC:ErrPulsarBrokerConfigNotFound"),
	)
	ErrPulsarTopicNotExists = errors.Normalize("pulsar topic not exists after creation",
		errors.RFCCodeText("CDC:ErrPulsarTopicNotExists"),
	)

	ErrRedoConfigInvalid = errors.Normalize(
		"redo log config invalid",
		errors.RFCCodeText("CDC:ErrRedoConfigInvalid"),
	)
	ErrRedoDownloadFailed = errors.Normalize(
		"redo log down load to local failed",
		errors.RFCCodeText("CDC:ErrRedoDownloadFailed"),
	)
	ErrRedoWriterStopped = errors.Normalize(
		"redo log writer stopped",
		errors.RFCCodeText("CDC:ErrRedoWriterStopped"),
	)
	ErrRedoFileOp = errors.Normalize(
		"redo file operation",
		errors.RFCCodeText("CDC:ErrRedoFileOp"),
	)
	ErrRedoMetaFileNotFound = errors.Normalize(
		"no redo meta file found in dir: %s",
		errors.RFCCodeText("CDC:ErrRedoMetaFileNotFound"),
	)
	ErrRedoMetaInitialize = errors.Normalize(
		"initialize meta for redo log",
		errors.RFCCodeText("CDC:ErrRedoMetaInitialize"),
	)
	ErrFileSizeExceed = errors.Normalize(
		"rawData size %d exceeds maximum file size %d",
		errors.RFCCodeText("CDC:ErrFileSizeExceed"),
	)
	ErrExternalStorageAPI = errors.Normalize(
		"external storage api",
		errors.RFCCodeText("CDC:ErrS3StorageAPI"),
	)
	ErrStorageInitialize = errors.Normalize(
		"fail to open storage for redo log",
		errors.RFCCodeText("CDC:ErrStorageInitialize"),
	)
	ErrCodecInvalidConfig = errors.Normalize(
		"Codec invalid config",
		errors.RFCCodeText("CDC:ErrCodecInvalidConfig"),
	)

	ErrCompressionFailed = errors.Normalize(
		"Compression failed",
		errors.RFCCodeText("CDC:ErrCompressionFailed"),
	)

	ErrSinkURIInvalid = errors.Normalize(
		"sink uri invalid '%s'",
		errors.RFCCodeText("CDC:ErrSinkURIInvalid"),
	)
	ErrIncompatibleSinkConfig = errors.Normalize(
		"incompatible configuration in sink uri(%s) and config file(%s), "+
			"please try to update the configuration only through sink uri",
		errors.RFCCodeText("CDC:ErrIncompatibleSinkConfig"),
	)
	ErrSinkUnknownProtocol = errors.Normalize(
		"unknown '%s' message protocol for sink",
		errors.RFCCodeText("CDC:ErrSinkUnknownProtocol"),
	)
	ErrMySQLTxnError = errors.Normalize(
		"MySQL txn error",
		errors.RFCCodeText("CDC:ErrMySQLTxnError"),
	)
	ErrMySQLDuplicateEntry = errors.Normalize(
		"MySQL duplicate entry error",
		errors.RFCCodeText("CDC:ErrMySQLDuplicateEntry"),
	)
	ErrMySQLQueryError = errors.Normalize(
		"MySQL query error",
		errors.RFCCodeText("CDC:ErrMySQLQueryError"),
	)
	ErrMySQLConnectionError = errors.Normalize(
		"MySQL connection error",
		errors.RFCCodeText("CDC:ErrMySQLConnectionError"),
	)
	ErrMySQLInvalidConfig = errors.Normalize(
		"MySQL config invalid",
		errors.RFCCodeText("CDC:ErrMySQLInvalidConfig"),
	)
	ErrMySQLWorkerPanic = errors.Normalize(
		"MySQL worker panic",
		errors.RFCCodeText("CDC:ErrMySQLWorkerPanic"),
	)
	ErrAvroToEnvelopeError = errors.Normalize(
		"to envelope failed",
		errors.RFCCodeText("CDC:ErrAvroToEnvelopeError"),
	)
	ErrAvroMarshalFailed = errors.Normalize(
		"json marshal failed",
		errors.RFCCodeText("CDC:ErrAvroMarshalFailed"),
	)
	ErrAvroEncodeFailed = errors.Normalize(
		"encode to avro native data",
		errors.RFCCodeText("CDC:ErrAvroEncodeFailed"),
	)
	ErrAvroEncodeToBinary = errors.Normalize(
		"encode to binray from native",
		errors.RFCCodeText("CDC:ErrAvroEncodeToBinary"),
	)
	ErrAvroSchemaAPIError = errors.Normalize(
		"schema manager API error, %s",
		errors.RFCCodeText("CDC:ErrAvroSchemaAPIError"),
	)
	ErrAvroInvalidMessage = errors.Normalize(
		"avro invalid message format, %s",
		errors.RFCCodeText("CDC:ErrAvroInvalidMessage"),
	)
	ErrMaxwellEncodeFailed = errors.Normalize(
		"maxwell encode failed",
		errors.RFCCodeText("CDC:ErrMaxwellEncodeFailed"),
	)
	ErrMaxwellInvalidData = errors.Normalize(
		"maxwell invalid data",
		errors.RFCCodeText("CDC:ErrMaxwellInvalidData"),
	)
	ErrOpenProtocolCodecInvalidData = errors.Normalize(
		"open-protocol codec invalid data",
		errors.RFCCodeText("CDC:ErrOpenProtocolCodecInvalidData"),
	)
	ErrCanalDecodeFailed = errors.Normalize(
		"canal decode failed",
		errors.RFCCodeText("CDC:ErrCanalDecodeFailed"),
	)
	ErrCanalEncodeFailed = errors.Normalize(
		"canal encode failed",
		errors.RFCCodeText("CDC:ErrCanalEncodeFailed"),
	)
	ErrOldValueNotEnabled = errors.Normalize(
		"old value is not enabled",
		errors.RFCCodeText("CDC:ErrOldValueNotEnabled"),
	)
	ErrSinkInvalidConfig = errors.Normalize(
		"sink config invalid",
		errors.RFCCodeText("CDC:ErrSinkInvalidConfig"),
	)
	ErrSinkIncompatibleConfig = errors.Normalize(
		"incompatible configuration %s",
		errors.RFCCodeText("CDC:ErrSinkIncompatibleConfig"),
	)
	ErrCraftCodecInvalidData = errors.Normalize(
		"craft codec invalid data",
		errors.RFCCodeText("CDC:ErrCraftCodecInvalidData"),
	)
	ErrMessageTooLarge = errors.Normalize(
		"message is too large",
		errors.RFCCodeText("CDC:ErrMessageTooLarge"),
	)
	ErrStorageSinkInvalidDateSeparator = errors.Normalize(
		"date separator in storage sink is invalid",
		errors.RFCCodeText("CDC:ErrStorageSinkInvalidDateSeparator"),
	)
	ErrCSVEncodeFailed = errors.Normalize(
		"csv encode failed",
		errors.RFCCodeText("CDC:ErrCSVEncodeFailed"),
	)
	ErrCSVDecodeFailed = errors.Normalize(
		"csv decode failed",
		errors.RFCCodeText("CDC:ErrCSVDecodeFailed"),
	)
	ErrDebeziumEncodeFailed = errors.Normalize(
		"debezium encode failed",
		errors.RFCCodeText("CDC:ErrDebeziumEncodeFailed"),
	)
	ErrStorageSinkInvalidConfig = errors.Normalize(
		"storage sink config invalid",
		errors.RFCCodeText("CDC:ErrStorageSinkInvalidConfig"),
	)
	ErrStorageSinkInvalidFileName = errors.Normalize(
		"filename in storage sink is invalid",
		errors.RFCCodeText("CDC:ErrStorageSinkInvalidFileName"),
	)

	// utilities related errors
	ErrToTLSConfigFailed = errors.Normalize(
		"generate tls config failed",
		errors.RFCCodeText("CDC:ErrToTLSConfigFailed"),
	)
	ErrCheckClusterVersionFromPD = errors.Normalize(
		"failed to request PD %s, please try again later",
		errors.RFCCodeText("CDC:ErrCheckClusterVersionFromPD"),
	)
	ErrNewSemVersion = errors.Normalize(
		"create sem version",
		errors.RFCCodeText("CDC:ErrNewSemVersion"),
	)
	ErrCheckDirWritable = errors.Normalize(
		"check dir writable failed",
		errors.RFCCodeText("CDC:ErrCheckDirWritable"),
	)
	ErrCheckDirValid = errors.Normalize(
		"check dir valid failed",
		errors.RFCCodeText("CDC:ErrCheckDirValid"),
	)
	ErrGetDiskInfo = errors.Normalize(
		"get dir disk info failed",
		errors.RFCCodeText("CDC:ErrGetDiskInfo"),
	)
	ErrLoadTimezone = errors.Normalize(
		"load timezone",
		errors.RFCCodeText("CDC:ErrLoadTimezone"),
	)
	ErrURLFormatInvalid = errors.Normalize(
		"url format is invalid",
		errors.RFCCodeText("CDC:ErrURLFormatInvalid"),
	)
	ErrIntersectNoOverlap = errors.Normalize(
		"span doesn't overlap: %+v vs %+v",
		errors.RFCCodeText("CDC:ErrIntersectNoOverlap"),
	)
	ErrOperateOnClosedNotifier = errors.Normalize(
		"operate on a closed notifier",
		errors.RFCCodeText("CDC:ErrOperateOnClosedNotifier"),
	)
	ErrDiskFull = errors.Normalize(
		"failed to preallocate file because disk is full",
		errors.RFCCodeText("CDC:ErrDiskFull"))
	ErrWaitFreeMemoryTimeout = errors.Normalize(
		"wait free memory timeout",
		errors.RFCCodeText("CDC:ErrWaitFreeMemoryTimeout"),
	)

	// encode/decode, data format and data integrity errors
	ErrInvalidRecordKey = errors.Normalize(
		"invalid record key - %q",
		errors.RFCCodeText("CDC:ErrInvalidRecordKey"),
	)
	ErrCodecDecode = errors.Normalize(
		"codec decode error",
		errors.RFCCodeText("CDC:ErrCodecDecode"),
	)
	ErrUnknownMetaType = errors.Normalize(
		"unknown meta type %v",
		errors.RFCCodeText("CDC:ErrUnknownMetaType"),
	)
	ErrDatumUnflatten = errors.Normalize(
		"unflatten datume data",
		errors.RFCCodeText("CDC:ErrDatumUnflatten"),
	)
	ErrDecodeRowToDatum = errors.Normalize(
		"decode row data to datum failed",
		errors.RFCCodeText("CDC:ErrDecodeRowToDatum"),
	)
	ErrMarshalFailed = errors.Normalize(
		"marshal failed",
		errors.RFCCodeText("CDC:ErrMarshalFailed"),
	)
	ErrUnmarshalFailed = errors.Normalize(
		"unmarshal failed",
		errors.RFCCodeText("CDC:ErrUnmarshalFailed"),
	)
	ErrInvalidChangefeedID = errors.Normalize(
		fmt.Sprintf("%s, %s, %s, %s,",
			"bad changefeed id",
			`please match the pattern "^[a-zA-Z0-9]+(\-[a-zA-Z0-9]+)*$"`,
			"the length should no more than %d",
			`eg, "simple-changefeed-task"`),
		errors.RFCCodeText("CDC:ErrInvalidChangefeedID"),
	)
	ErrInvalidNamespace = errors.Normalize(
		fmt.Sprintf("%s, %s, %s, %s,",
			"bad namespace",
			`please match the pattern "^[a-zA-Z0-9]+(\-[a-zA-Z0-9]+)*$"`,
			"the length should no more than %d",
			`eg, "simple-namespace-test"`),
		errors.RFCCodeText("CDC:ErrInvalidNamespace"),
	)
	ErrInvalidEtcdKey = errors.Normalize(
		"invalid key: %s",
		errors.RFCCodeText("CDC:ErrInvalidEtcdKey"),
	)

	// schema storage errors
	ErrSchemaStorageUnresolved = errors.Normalize(
		"can not found schema snapshot, the specified ts(%d) is more than resolvedTs(%d)",
		errors.RFCCodeText("CDC:ErrSchemaStorageUnresolved"),
	)
	ErrSchemaStorageGCed = errors.Normalize(
		"can not found schema snapshot, the specified ts(%d) is less than gcTS(%d)",
		errors.RFCCodeText("CDC:ErrSchemaStorageGCed"),
	)
	ErrSchemaSnapshotNotFound = errors.Normalize(
		"can not found schema snapshot, ts: %d",
		errors.RFCCodeText("CDC:ErrSchemaSnapshotNotFound"),
	)
	ErrSchemaStorageTableMiss = errors.Normalize(
		"table %d not found",
		errors.RFCCodeText("CDC:ErrSchemaStorageTableMiss"),
	)
	ErrSnapshotSchemaNotFound = errors.Normalize(
		"schema %d not found in schema snapshot",
		errors.RFCCodeText("CDC:ErrSnapshotSchemaNotFound"),
	)
	ErrSnapshotTableNotFound = errors.Normalize(
		"table %d not found in schema snapshot",
		errors.RFCCodeText("CDC:ErrSnapshotTableNotFound"),
	)
	ErrSnapshotSchemaExists = errors.Normalize(
		"schema %s(%d) already exists",
		errors.RFCCodeText("CDC:ErrSnapshotSchemaExists"),
	)
	ErrSnapshotTableExists = errors.Normalize(
		"table %s.%s already exists",
		errors.RFCCodeText("CDC:ErrSnapshotTableExists"),
	)
	ErrInvalidDDLJob = errors.Normalize(
		"invalid ddl job(%d)",
		errors.RFCCodeText("CDC:ErrInvalidDDLJob"),
	)
	ErrExchangePartition = errors.Normalize(
		"exchange partition failed, %s",
		errors.RFCCodeText("CDC:ErrExchangePartition"),
	)

	ErrCorruptedDataMutation = errors.Normalize(
		"Changefeed %s.%s stopped due to corrupted data mutation received",
		errors.RFCCodeText("CDC:ErrCorruptedDataMutation"))

	// server related errors
	ErrCaptureSuicide = errors.Normalize(
		"capture suicide",
		errors.RFCCodeText("CDC:ErrCaptureSuicide"),
	)
	ErrCaptureRegister = errors.Normalize(
		"capture register to etcd failed",
		errors.RFCCodeText("CDC:ErrCaptureRegister"),
	)
	ErrCaptureNotInitialized = errors.Normalize(
		"capture has not been initialized yet",
		errors.RFCCodeText("CDC:ErrCaptureNotInitialized"),
	)
	ErrProcessorUnknown = errors.Normalize(
		"processor running unknown error",
		errors.RFCCodeText("CDC:ErrProcessorUnknown"),
	)
	ErrOwnerUnknown = errors.Normalize(
		"owner running unknown error",
		errors.RFCCodeText("CDC:ErrOwnerUnknown"),
	)
	ErrProcessorTableNotFound = errors.Normalize(
		"table not found in processor cache",
		errors.RFCCodeText("CDC:ErrProcessorTableNotFound"),
	)
	ErrInvalidServerOption = errors.Normalize(
		"invalid server option",
		errors.RFCCodeText("CDC:ErrInvalidServerOption"),
	)
	ErrServeHTTP = errors.Normalize(
		"serve http error",
		errors.RFCCodeText("CDC:ErrServeHTTP"),
	)
	ErrCaptureCampaignOwner = errors.Normalize(
		"campaign owner failed",
		errors.RFCCodeText("CDC:ErrCaptureCampaignOwner"),
	)
	ErrCaptureResignOwner = errors.Normalize(
		"resign owner failed",
		errors.RFCCodeText("CDC:ErrCaptureResignOwner"),
	)
	ErrClusterIsUnhealthy = errors.Normalize(
		"TiCDC cluster is unhealthy",
		errors.RFCCodeText("CDC:ErrClusterIsUnhealthy"),
	)
	ErrAPIInvalidParam = errors.Normalize(
		"invalid api parameter",
		errors.RFCCodeText("CDC:ErrAPIInvalidParam"),
	)
	ErrAPIGetPDClientFailed = errors.Normalize(
		"failed to get PDClient to connect PD, please recheck",
		errors.RFCCodeText("CDC:ErrAPIGetPDClientFailed"),
	)
	ErrRequestForwardErr = errors.Normalize(
		"request forward error, an request can only forward to owner one time",
		errors.RFCCodeText("ErrRequestForwardErr"),
	)
	ErrInternalServerError = errors.Normalize(
		"internal server error",
		errors.RFCCodeText("CDC:ErrInternalServerError"),
	)
	ErrChangefeedUpdateRefused = errors.Normalize(
		"changefeed update error: %s",
		errors.RFCCodeText("CDC:ErrChangefeedUpdateRefused"),
	)
	ErrChangefeedUpdateFailedTransaction = errors.Normalize(
		"changefeed update failed due to unexpected etcd transaction failure: %s",
		errors.RFCCodeText("CDC:ErrChangefeedUpdateFailed"),
	)
	ErrUpdateServiceSafepointFailed = errors.Normalize(
		"updating service safepoint failed",
		errors.RFCCodeText("CDC:ErrUpdateServiceSafepointFailed"),
	)
	ErrStartTsBeforeGC = errors.Normalize(
		"fail to create or maintain changefeed because start-ts %d "+
			"is earlier than or equal to GC safepoint at %d",
		errors.RFCCodeText("CDC:ErrStartTsBeforeGC"),
	)
	ErrTargetTsBeforeStartTs = errors.Normalize(
		"fail to create changefeed because target-ts %d is earlier than start-ts %d",
		errors.RFCCodeText("CDC:ErrTargetTsBeforeStartTs"),
	)
	ErrSnapshotLostByGC = errors.Normalize(
		"fail to create or maintain changefeed due to snapshot loss"+
			" caused by GC. checkpoint-ts %d is earlier than or equal to GC safepoint at %d",
		errors.RFCCodeText("CDC:ErrSnapshotLostByGC"),
	)
	ErrGCTTLExceeded = errors.Normalize(
		"the checkpoint-ts(%d) lag of the changefeed(%s) has exceeded "+
			"the GC TTL and the changefeed is blocking global GC progression",
		errors.RFCCodeText("CDC:ErrGCTTLExceeded"),
	)
	ErrNotOwner = errors.Normalize(
		"this capture is not a owner",
		errors.RFCCodeText("CDC:ErrNotOwner"),
	)
	ErrOwnerNotFound = errors.Normalize(
		"owner not found",
		errors.RFCCodeText("CDC:ErrOwnerNotFound"),
	)
	ErrTableIneligible = errors.Normalize(
		"some tables are not eligible to replicate(%v), "+
			"if you want to ignore these tables, please set ignore_ineligible_table to true",
		errors.RFCCodeText("CDC:ErrTableIneligible"),
	)
	ErrInvalidCheckpointTs = errors.Normalize(
		"checkpointTs(%v) should not larger than resolvedTs(%v)",
		errors.RFCCodeText("CDC:ErrInvalidCheckpointTs"),
	)

	// EtcdWorker related errors. Internal use only.
	// ErrEtcdTryAgain is used by a PatchFunc to force a transaction abort.
	ErrEtcdTryAgain = errors.Normalize(
		"the etcd txn should be aborted and retried immediately",
		errors.RFCCodeText("CDC:ErrEtcdTryAgain"),
	)
	// ErrEtcdIgnore is used by a PatchFunc to signal that the reactor no longer wishes to update Etcd.
	ErrEtcdIgnore = errors.Normalize(
		"this patch should be excluded from the current etcd txn",
		errors.RFCCodeText("CDC:ErrEtcdIgnore"),
	)
	// ErrEtcdSessionDone is used by etcd worker to signal a session done
	ErrEtcdSessionDone = errors.Normalize(
		"the etcd session is done",
		errors.RFCCodeText("CDC:ErrEtcdSessionDone"),
	)
	// ErrReactorFinished is used by reactor to signal a **normal** exit.
	ErrReactorFinished = errors.Normalize(
		"the reactor has done its job and should no longer be executed",
		errors.RFCCodeText("CDC:ErrReactorFinished"),
	)
	ErrLeaseExpired = errors.Normalize(
		"owner lease expired ",
		errors.RFCCodeText("CDC:ErrLeaseExpired"),
	)
	ErrEtcdTxnSizeExceed = errors.Normalize(
		"patch size:%d of a single changefeed exceed etcd txn max size:%d",
		errors.RFCCodeText("CDC:ErrEtcdTxnSizeExceed"),
	)
	ErrEtcdTxnOpsExceed = errors.Normalize(
		"patch ops:%d of a single changefeed exceed etcd txn max ops:%d",
		errors.RFCCodeText("CDC:ErrEtcdTxnOpsExceed"),
	)
	ErrEtcdMigrateFailed = errors.Normalize(
		"etcd meta data migrate failed:%s",
		errors.RFCCodeText("CDC:ErrEtcdMigrateFailed"),
	)
	ErrChangefeedUnretryable = errors.Normalize(
		"changefeed is in unretryable state, please check the error message"+
			", and you should manually handle it",
		errors.RFCCodeText("CDC:ErrChangefeedUnretryable"),
	)

	// workerpool errors
	ErrWorkerPoolHandleCancelled = errors.Normalize(
		"workerpool handle is cancelled",
		errors.RFCCodeText("CDC:ErrWorkerPoolHandleCancelled"),
	)
	ErrAsyncPoolExited = errors.Normalize(
		"asyncPool has exited. Report a bug if seen externally.",
		errors.RFCCodeText("CDC:ErrAsyncPoolExited"),
	)
	ErrWorkerPoolGracefulUnregisterTimedOut = errors.Normalize(
		"workerpool handle graceful unregister timed out",
		errors.RFCCodeText("CDC:ErrWorkerPoolGracefulUnregisterTimedOut"),
	)

	// redo log related errors
	ErrConsistentStorage = errors.Normalize(
		"consistent storage (%s) not support",
		errors.RFCCodeText("CDC:ErrConsistentStorage"),
	)
	// sorter errors
	ErrIllegalSorterParameter = errors.Normalize(
		"illegal parameter for sorter: %s",
		errors.RFCCodeText("CDC:ErrIllegalSorterParameter"),
	)
	ErrConflictingFileLocks = errors.Normalize(
		"file lock conflict: %s",
		errors.RFCCodeText("ErrConflictingFileLocks"),
	)

	// retry error
	ErrReachMaxTry = errors.Normalize("reach maximum try: %s, error: %s",
		errors.RFCCodeText("CDC:ErrReachMaxTry"),
	)

	// tcp server error
	ErrTCPServerClosed = errors.Normalize("The TCP server has been closed",
		errors.RFCCodeText("CDC:ErrTCPServerClosed"),
	)

	// p2p error
	ErrPeerMessageIllegalMeta = errors.Normalize(
		"peer-to-peer message server received an RPC call with illegal metadata",
		errors.RFCCodeText("CDC:ErrPeerMessageIllegalMeta"),
	)
	ErrPeerMessageClientPermanentFail = errors.Normalize(
		"peer-to-peer message client has failed permanently, no need to reconnect: %s",
		errors.RFCCodeText("CDC:ErrPeerMessageClientPermanentFail"),
	)
	ErrPeerMessageClientClosed = errors.Normalize(
		"peer-to-peer message client has been closed",
		errors.RFCCodeText("CDC:ErrPeerMessageClientClosed"),
	)
	ErrPeerMessageSendTryAgain = errors.Normalize(
		"peer-to-peer message client has too many pending messages to send,"+
			" try again later",
		errors.RFCCodeText("CDC:ErrPeerMessageSendTryAgain"),
	)
	ErrPeerMessageEncodeError = errors.Normalize(
		"failed to encode peer-to-peer message",
		errors.RFCCodeText("CDC:ErrPeerMessageEncodeError"),
	)
	ErrPeerMessageInternalSenderClosed = errors.Normalize(
		"peer-to-peer message server tries to send to a closed stream. Internal only.",
		errors.RFCCodeText("CDC:ErrPeerMessageInternalSenderClosed"),
	)
	ErrPeerMessageStaleConnection = errors.Normalize(
		"peer-to-peer message stale connection: old-epoch %d, new-epoch %d",
		errors.RFCCodeText("CDC:ErrPeerMessageStaleConnection"),
	)
	ErrPeerMessageDuplicateConnection = errors.Normalize(
		"peer-to-peer message duplicate connection: epoch %d",
		errors.RFCCodeText("CDC:ErrPeerMessageDuplicateConnection"),
	)
	ErrPeerMessageServerClosed = errors.Normalize(
		"peer-to-peer message server has closed connection: %s.",
		errors.RFCCodeText("CDC:ErrPeerMessageServerClosed"),
	)
	ErrPeerMessageDataLost = errors.Normalize(
		"peer-to-peer message data lost, topic: %s, seq: %d",
		errors.RFCCodeText("CDC:ErrPeerMessageDataLost"),
	)
	ErrPeerMessageToManyPeers = errors.Normalize(
		"peer-to-peer message server got too many peers: %d peers",
		errors.RFCCodeText("CDC:ErrPeerMessageToManyPeers"),
	)
	ErrPeerMessageDecodeError = errors.Normalize(
		"failed to decode peer-to-peer message",
		errors.RFCCodeText("CDC:ErrPeerMessageDecodeError"),
	)
	ErrPeerMessageTaskQueueCongested = errors.Normalize(
		"peer-to-peer message server has too many pending tasks",
		errors.RFCCodeText("CDC:ErrPeerMessageTaskQueueCongested"),
	)
	ErrPeerMessageReceiverMismatch = errors.Normalize(
		"peer-to-peer message receiver is a mismatch: expected %s, got %s",
		errors.RFCCodeText("CDC:ErrPeerMessageReceiverMismatch"),
	)
	ErrPeerMessageTopicCongested = errors.Normalize(
		"peer-to-peer message topic has congested, aborting all connections",
		errors.RFCCodeText("CDC:ErrPeerMessageTopicCongested"),
	)
	ErrPeerMessageInjectedServerRestart = errors.Normalize(
		"peer-to-peer message server injected error",
		errors.RFCCodeText("CDC:ErrPeerMessageInjectedServerRestart"),
	)

	// RESTful client error
	ErrRewindRequestBodyError = errors.Normalize(
		"failed to seek to the beginning of request body",
		errors.RFCCodeText("CDC:ErrRewindRequestBodyError"),
	)
	ErrZeroLengthResponseBody = errors.Normalize(
		"0-length response with status code: %d",
		errors.RFCCodeText("CDC:ErrZeroLengthResponseBody"),
	)
	ErrInvalidHost = errors.Normalize(
		"host must be a URL or a host:port pair: %q",
		errors.RFCCodeText("CDC:ErrInvalidHost"),
	)

	// Upstream error
	ErrUpstreamNotFound = errors.Normalize(
		"upstream not found, cluster-id: %d",
		errors.RFCCodeText("CDC:ErrUpstreamNotFound"),
	)
	ErrUpstreamManagerNotReady = errors.Normalize(
		"upstream manager not ready",
		errors.RFCCodeText("CDC:ErrUpstreamManagerNotReady"),
	)
	ErrUpstreamClosed = errors.Normalize(
		"upstream has been closed",
		errors.RFCCodeText("CDC:ErrUpstreamClosed"),
	)
	ErrUpstreamHasRunningImport = errors.Normalize(
		"upstream has running import tasks, upstream-id: %d",
		errors.RFCCodeText("CDC:ErrUpstreamHasRunningImport"),
	)

	// ReplicationSet error
	ErrReplicationSetInconsistent = errors.Normalize(
		"replication set inconsistent: %s",
		errors.RFCCodeText("CDC:ErrReplicationSetInconsistent"),
	)
	ErrReplicationSetMultiplePrimaryError = errors.Normalize(
		"replication set multiple primary: %s",
		errors.RFCCodeText("CDC:ErrReplicationSetMultiplePrimaryError"),
	)

	ErrUpstreamMissMatch = errors.Normalize(
		"upstream missmatch,old: %d, new %d",
		errors.RFCCodeText("CDC:ErrUpstreamMissMatch"),
	)

	ErrServerIsNotReady = errors.Normalize(
		"cdc server is not ready",
		errors.RFCCodeText("CDC:ErrServerIsNotReady"),
	)

	// cli error
	ErrCliInvalidCheckpointTs = errors.Normalize(
		"invalid overwrite-checkpoint-ts %s, "+
			"overwrite-checkpoint-ts only accept 'now' or a valid timestamp in integer",
		errors.RFCCodeText("CDC:ErrCliInvalidCheckpointTs"),
	)
	ErrCliCheckpointTsIsInFuture = errors.Normalize(
		"the overwrite-checkpoint-ts %d must be smaller than current TSO",
		errors.RFCCodeText("CDC:ErrCliCheckpointTsIsInFuture"),
	)
	ErrCliAborted = errors.Normalize(
		"command '%s' is aborted by user",
		errors.RFCCodeText("CDC:ErrCliAborted"),
	)
	// Filter error
	ErrFailedToFilterDML = errors.Normalize(
		"failed to filter dml event: %v, please report a bug",
		errors.RFCCodeText("CDC:ErrFailedToFilterDML"),
	)
	ErrExpressionParseFailed = errors.Normalize(
		"invalid filter expressions. There is a syntax error in: '%s'",
		errors.RFCCodeText("CDC:ErrInvalidFilterExpression"),
	)
	ErrExpressionColumnNotFound = errors.Normalize(
		"invalid filter expression(s). Cannot find column '%s' from table '%s' in: %s",
		errors.RFCCodeText("CDC:ErrExpressionColumnNotFound"),
	)
	ErrInvalidIgnoreEventType = errors.Normalize(
		"invalid ignore event type: '%s'",
		errors.RFCCodeText("CDC:ErrInvalidIgnoreEventType"),
	)
	ErrConvertDDLToEventTypeFailed = errors.Normalize(
		"failed to convert ddl '%s' to filter event type",
		errors.RFCCodeText("CDC:ErrConvertDDLToEventTypeFailed"),
	)
	ErrSyncRenameTableFailed = errors.Normalize(
		"table's old name is not in filter rule, and its new name in filter rule "+
			"table id '%d', ddl query: [%s], it's an unexpected behavior, "+
			"if you want to replicate this table, please add its old name to filter rule.",
		errors.RFCCodeText("CDC:ErrSyncRenameTableFailed"),
	)

	// changefeed config error
	ErrInvalidReplicaConfig = errors.Normalize(
		"invalid replica config, %s",
		errors.RFCCodeText("CDC:ErrInvalidReplicaConfig"),
	)
	ErrInternalCheckFailed = errors.Normalize(
		"internal check failed, %s",
		errors.RFCCodeText("CDC:ErrInternalCheckFailed"),
	)

	ErrHandleDDLFailed = errors.Normalize(
		"handle ddl failed, query: %s, startTs: %d. "+
			"If you want to skip this DDL and continue with replication, "+
			"you can manually execute this DDL downstream. Afterwards, "+
			"add `ignore-txn-start-ts=[%d]` to the changefeed in the filter configuration.",
		errors.RFCCodeText("CDC:ErrHandleDDLFailed"),
	)

	ErrInvalidGlueSchemaRegistryConfig = errors.Normalize(
		"invalid glue schema registry config, %s",
		errors.RFCCodeText("CDC:ErrInvalidGlueSchemaRegistryConfig"),
	)

	// cdc v2
	// TODO(CharlesCheung): refactor this errors
	ErrElectorNotLeader = errors.Normalize(
		"%s is not leader",
		errors.RFCCodeText("CDC:ErrNotLeader"),
	)
	ErrNotController = errors.Normalize(
		"not controller",
		errors.RFCCodeText("CDC:ErrNotController"),
	)
	ErrMetaRowsAffectedNotMatch = errors.Normalize(
		"rows affected by the operation %s is unexpected: expected %d, got %d",
		errors.RFCCodeText("CDC:ErrMetaOpIgnored"),
	)
	ErrMetaOpFailed = errors.Normalize(
		"unexpected meta operation failure: %s",
		errors.RFCCodeText("DFLOW:ErrMetaOpFailed"),
	)
	ErrMetaInvalidState = errors.Normalize(
		"meta state is invalid: %s",
		errors.RFCCodeText("DFLOW:ErrMetaInvalidState"),
	)
	ErrInconsistentMetaCache = errors.Normalize(
		"meta cache is inconsistent: %s",
		errors.RFCCodeText("DFLOW:ErrInconsistentMetaCache"),
	)

	ErrUnexpected = errors.Normalize(
		"cdc met unexpected error: %s",
		errors.RFCCodeText("CDC:ErrUnexpected"),
	)

	// credential related errors
	ErrCredentialNotFound = errors.Normalize(
		"credential not found: %s",
		errors.RFCCodeText("CDC:ErrCredentialNotFound"),
	)
	ErrUnauthorized = errors.Normalize(
		"user %s unauthorized, error: %s",
		errors.RFCCodeText("CDC:ErrUnauthorized"),
	)
)

errors

var (
	// general errors
	ErrUnknown = errors.Normalize(
		"unknown error",
		errors.RFCCodeText("DFLOW:ErrUnknown"),
	)
	ErrInvalidArgument = errors.Normalize(
		"invalid argument: %s",
		errors.RFCCodeText("DFLOW:ErrInvalidArgument"),
	)

	ErrClusterResourceNotEnough = errors.Normalize(
		"cluster resource is not enough, please scale out the cluster",
		errors.RFCCodeText("DFLOW:ErrClusterResourceNotEnough"),
	)

	ErrDecodeEtcdKeyFail = errors.Normalize(
		"failed to decode etcd key: %s",
		errors.RFCCodeText("DFLOW:ErrDecodeEtcdKeyFail"),
	)
	ErrEtcdAPIError = errors.Normalize(
		"etcd api returns error",
		errors.RFCCodeText("DFLOW:ErrEtcdAPIError"),
	)

	// master related errors
	ErrMasterNotReady = errors.Normalize(
		"master is not ready",
		errors.RFCCodeText("DFLOW:ErrMasterNotReady"),
	)
	ErrMasterNoLeader = errors.Normalize(
		"master has no leader",
		errors.RFCCodeText("DFLOW:ErrMasterNoLeader"),
	)
	ErrMasterDecodeConfigFile = errors.Normalize(
		"decode config file failed",
		errors.RFCCodeText("DFLOW:ErrMasterDecodeConfigFile"),
	)
	ErrMasterConfigUnknownItem = errors.Normalize(
		"master config contains unknown configuration options: %s",
		errors.RFCCodeText("DFLOW:ErrMasterConfigUnknownItem"),
	)
	ErrMasterNotFound = errors.Normalize(
		"master is not found: master ID %s",
		errors.RFCCodeText("DFLOW:ErrMasterNotFound"),
	)
	ErrMasterClosed = errors.Normalize(
		"master has been closed explicitly: master ID %s",
		errors.RFCCodeText("DFLOW:ErrMasterClosed"),
	)
	ErrMasterConcurrencyExceeded = errors.Normalize(
		"master has reached concurrency quota",
		errors.RFCCodeText("DFLOW:ErrMasterConcurrencyExceeded"),
	)
	ErrMasterCreateWorkerBackoff = errors.Normalize(
		"create worker is being backoff, retry later",
		errors.RFCCodeText("DFLOW:ErrMasterCreateWorkerBackoff"),
	)
	ErrMasterCreateWorkerTerminate = errors.Normalize(
		"create worker is terminated, won't backoff any more",
		errors.RFCCodeText("DFLOW:ErrMasterCreateWorkerTerminate"),
	)
	ErrMasterInvalidMeta = errors.Normalize(
		"invalid master meta data: %s",
		errors.RFCCodeText("DFLOW:ErrMasterInvalidMeta"),
	)
	ErrInvalidMasterMessage = errors.Normalize(
		"invalid master message: %s",
		errors.RFCCodeText("DFLOW:ErrInvalidMasterMessage"),
	)
	ErrSendingMessageToTombstone = errors.Normalize(
		"trying to send message to a tombstone worker handle: %s",
		errors.RFCCodeText("DFLOW:ErrSendingMessageToTombstone"),
	)
	ErrMasterInterfaceNotFound = errors.Normalize(
		"basemaster interface not found",
		errors.RFCCodeText("DFLOW:ErrMasterInterfaceNotFound"),
	)
	ErrExecutorWatcherClosed = errors.Normalize(
		"executor watcher is closed",
		errors.RFCCodeText("DFLOW:ErrExecutorWatcherClosed"),
	)

	ErrWorkerTypeNotFound = errors.Normalize(
		"worker type is not found: type %d",
		errors.RFCCodeText("DFLOW:ErrWorkerTypeNotFound"),
	)
	ErrWorkerNotFound = errors.Normalize(
		"worker is not found: worker ID %s",
		errors.RFCCodeText("DFLOW:ErrWorkerNotFound"),
	)
	ErrWorkerOffline = errors.Normalize(
		"worker is offline: workerID: %s, error message: %s",
		errors.RFCCodeText("DFLOW:ErrWorkerOffline"),
	)
	ErrWorkerSuicide = errors.Normalize(
		"worker has committed suicide due to master(%s) having timed out",
		errors.RFCCodeText("DFLOW:ErrWorkerSuicide"),
	)
	ErrWorkerUpdateStatusTryAgain = errors.Normalize(
		"worker should try again in updating the status",
		errors.RFCCodeText("DFLOW:ErrWorkerUpdateStatusTryAgain"),
	)
	// TODO: unify the following three errors into one ErrWorkerTerminated
	ErrWorkerFinish = errors.Normalize(
		"worker finished and exited",
		errors.RFCCodeText("DFLOW:ErrWorkerFinish"),
	)
	ErrWorkerCancel = errors.Normalize(
		"worker is canceled",
		errors.RFCCodeText("DFLOW:ErrWorkerCancel"),
	)
	ErrWorkerFailed = errors.Normalize(
		"worker is failed permanently",
		errors.RFCCodeText("DFLOW:ErrWorkerFailed"),
	)
	ErrTooManyStatusUpdates = errors.Normalize(
		"there are too many pending worker status updates: %d",
		errors.RFCCodeText("DFLOW:ErrTooManyStatusUpdates"),
	)
	ErrWorkerHalfExit = errors.Normalize(
		"the worker is in half-exited state",
		errors.RFCCodeText("DFLOW:ErrWorkerHalfExit"),
	)
	// ErrCreateWorkerNonTerminate indicates the job can be re-created.
	ErrCreateWorkerNonTerminate = errors.Normalize(
		"create worker is not terminated",
		errors.RFCCodeText("DFLOW:ErrCreateWorkerNonTerminate"),
	)
	// ErrCreateWorkerTerminate indicates the job should be terminated permanently
	// from the perspective of business logic.
	ErrCreateWorkerTerminate = errors.Normalize(
		"create worker is terminated",
		errors.RFCCodeText("DFLOW:ErrCreateWorkerTerminate"),
	)

	// job manager related errors
	ErrJobManagerGetJobDetailFail = errors.Normalize(
		"failed to get job detail from job master",
		errors.RFCCodeText("DFLOW:ErrJobManagerGetJobDetailFail"),
	)

	// master etcd related errors
	ErrMasterEtcdEpochFail = errors.Normalize(
		"server master generate epoch fail",
		errors.RFCCodeText("DFLOW:ErrMasterEtcdEpochFail"),
	)

	// executor related errors
	ErrUnknownExecutor = errors.Normalize(
		"unknown executor: %s",
		errors.RFCCodeText("DFLOW:ErrUnknownExecutor"),
	)
	ErrTombstoneExecutor = errors.Normalize(
		"tombstone executor: %s",
		errors.RFCCodeText("DFLOW:ErrTombstoneExecutor"),
	)
	ErrExecutorNotFound = errors.Normalize(
		"executor %s not found",
		errors.RFCCodeText("DFLOW:ErrExecutorNotFound"),
	)
	ErrExecutorAlreadyExists = errors.Normalize(
		"executor %s already exists",
		errors.RFCCodeText("DFLOW:ErrExecutorAlreadyExists"),
	)
	ErrExecutorDecodeConfigFile = errors.Normalize(
		"decode config file failed",
		errors.RFCCodeText("DFLOW:ErrExecutorDecodeConfigFile"),
	)
	ErrExecutorConfigUnknownItem = errors.Normalize(
		"master config contains unknown configuration options: %s",
		errors.RFCCodeText("DFLOW:ErrExecutorConfigUnknownItem"),
	)
	ErrHeartbeat = errors.Normalize(
		"heartbeat error type: %s",
		errors.RFCCodeText("DFLOW:ErrHeartbeat"),
	)
	ErrRuntimeIncomingQueueFull = errors.Normalize(
		"runtime has too many pending CreateWorker requests",
		errors.RFCCodeText("DFLOW:ErrRuntimeIncomingQueueFull"),
	)
	ErrRuntimeIsClosed = errors.Normalize(
		"runtime has been closed",
		errors.RFCCodeText("DFLOW:ErrRuntimeIsClosed"),
	)
	ErrRuntimeDuplicateTaskID = errors.Normalize(
		"trying to add a task with the same ID as an existing one",
		errors.RFCCodeText("DFLOW:ErrRuntimeDuplicateTaskID %s"),
	)
	ErrRuntimeClosed = errors.Normalize(
		"runtime has been closed",
		errors.RFCCodeText("DFLOW:ErrRuntimeClosed"),
	)
	ErrExecutorNotFoundForMessage = errors.Normalize(
		"cannot find the executor for p2p messaging",
		errors.RFCCodeText("DFLOW:ErrExecutorNotFoundForMessage"),
	)
	ErrMasterTooManyPendingEvents = errors.Normalize(
		"master has too many pending events",
		errors.RFCCodeText("DFLOW:ErrMasterTooManyPendingEvents"),
	)

	// meta related errors
	ErrMetaNewClientFail = errors.Normalize(
		"create meta client fail",
		errors.RFCCodeText("DFLOW:ErrMetaNewClientFail"),
	)
	ErrMetaOpFail = errors.Normalize(
		"meta operation fail",
		errors.RFCCodeText("DFLOW:ErrMetaOpFail"),
	)
	ErrMetaOptionInvalid = errors.Normalize(
		"meta option invalid",
		errors.RFCCodeText("DFLOW:ErrMetaOptionInvalid"),
	)
	ErrMetaOptionConflict = errors.Normalize(
		"WithRange/WithPrefix/WithFromKey, more than one option are used",
		errors.RFCCodeText("DFLOW:ErrMetaOptionConflict"),
	)
	ErrMetaEmptyKey = errors.Normalize(
		"meta empty key",
		errors.RFCCodeText("DFLOW:ErrMetaEmptyKey"),
	)
	ErrMetaNestedTxn = errors.Normalize(
		"meta unsupported nested txn",
		errors.RFCCodeText("DFLOW:ErrMetaNestedTxn"),
	)
	ErrMetaCommittedTxn = errors.Normalize(
		"meta already committed txn",
		errors.RFCCodeText("DFLOW:ErrMetaCommittedTxn"),
	)
	ErrMetaStoreIDDuplicate = errors.Normalize(
		"metastore id duplicated",
		errors.RFCCodeText("DFLOW:ErrMetaStoreIDDuplicate"),
	)
	ErrMetaEntryNotFound = errors.Normalize(
		"meta entry not found",
		errors.RFCCodeText("DFLOW:ErrMetaEntryNotFound"),
	)
	ErrMetaParamsInvalid = errors.Normalize(
		"meta params invalid:%s",
		errors.RFCCodeText("DFLOW:ErrMetaParamsInvalid"),
	)
	ErrMetaClientTypeNotSupport = errors.Normalize(
		"meta client type not support:%s",
		errors.RFCCodeText("DFLOW:ErrMetaClientTypeNotSupport"),
	)

	// Resource related errors
	ErrDuplicateResourceID = errors.Normalize(
		"duplicate resource ID: %s",
		errors.RFCCodeText("DFLOW:ErrDuplicateResourceID"),
	)
	ErrResourceAlreadyExists = errors.Normalize(
		"resource %s already exists",
		errors.RFCCodeText("DFLOW:ErrResourceAlreadyExists"),
	)
	ErrIllegalResourcePath = errors.Normalize(
		"resource path is illegal: %s",
		errors.RFCCodeText("DFLOW:ErrIllegalResourcePath"),
	)
	ErrResourceDoesNotExist = errors.Normalize(
		"resource does not exists: %s",
		errors.RFCCodeText("DFLOW:ErrResourceDoesNotExist"),
	)
	ErrResourceConflict = errors.Normalize(
		"resource % on executor %s conflicts with resource %s on executor %s",
		errors.RFCCodeText("DFLOW:ErrResourceConflict"),
	)
	ErrReadLocalFileDirectoryFailed = errors.Normalize(
		"reading local file resource directory failed",
		errors.RFCCodeText("DFLOW:ErrReadLocalFileDirectoryFailed"),
	)
	ErrCreateLocalFileDirectoryFailed = errors.Normalize(
		"creating local file resource directory failed",
		errors.RFCCodeText("DFLOW:ErrCreateLocalFileDirectoryFailed"),
	)
	ErrCleaningLocalTempFiles = errors.Normalize(
		"errors is encountered when cleaning local temp files",
		errors.RFCCodeText("DFLOW:ErrCleaningLocalTempFiles"),
	)
	ErrRemovingLocalResource = errors.Normalize(
		"removing a local resource file directory has failed",
		errors.RFCCodeText("DFLOW:ErrRemovingLocalResource"),
	)
	ErrFailToCreateExternalStorage = errors.Normalize(
		"failed to create external storage",
		errors.RFCCodeText("DFLOW:ErrFailToCreateExternalStorage"),
	)
	ErrInvalidResourceHandle = errors.Normalize(
		"using an invalid resource handle",
		errors.RFCCodeText("DFLOW:ErrInvalidResourceHandle"),
	)
	ErrLocalFileDirNotWritable = errors.Normalize(
		"local resource directory not writable",
		errors.RFCCodeText("DFLOW:ErrLocalFileDirNotWritable"),
	)
	ErrNoQualifiedExecutor = errors.Normalize(
		"no executor is available for scheduling",
		errors.RFCCodeText("DFLOW:ErrNoQualifiedExecutor"),
	)
	ErrFilterNoResult = errors.Normalize(
		"filter % returns no result",
		errors.RFCCodeText("DFLOW:ErrFilterNoResult"),
	)
	ErrSelectorUnsatisfied = errors.Normalize(
		"selector %v is not satisfied",
		errors.RFCCodeText("DFLOW:ErrSelectorUnsatisfied"),
	)
	ErrResourceFilesNotFound = errors.Normalize(
		"resource files not found",
		errors.RFCCodeText("DFLOW:ErrResourceFilesNotFound"),
	)
	ErrResourceMetastoreError = errors.Normalize(
		"resource metastore error",
		errors.RFCCodeText("DFLOW:ErrResourceMetastoreError"),
	)

	// Job related error
	ErrJobNotFound = errors.Normalize(
		"job %s is not found",
		errors.RFCCodeText("DFLOW:ErrJobNotFound"),
	)
	ErrJobAlreadyExists = errors.Normalize(
		"job %s already exists",
		errors.RFCCodeText("DFLOW:ErrJobAlreadyExists"),
	)
	ErrJobAlreadyCanceled = errors.Normalize(
		"job %s is already canceled",
		errors.RFCCodeText("DFLOW:ErrJobAlreadyCanceled"),
	)
	ErrJobNotTerminated = errors.Normalize(
		"job %s is not terminated",
		errors.RFCCodeText("DFLOW:ErrJobNotTerminated"),
	)
	ErrJobNotRunning = errors.Normalize(
		"job %s is not running",
		errors.RFCCodeText("DFLOW:ErrJobNotRunning"),
	)

	// metastore related errors
	ErrMetaStoreNotExists = errors.Normalize(
		"metastore %s does not exist",
		errors.RFCCodeText("DFLOW:ErrMetaStoreNotExists"),
	)

	// cli related errors
	ErrInvalidCliParameter = errors.Normalize(
		"invalid cli parameters",
		errors.RFCCodeText("DFLOW:ErrInvalidCliParameter"),
	)

	ErrIncompatibleSchedulerRequest = errors.Normalize(
		"incompatible scheduler request: %s",
		errors.RFCCodeText("DFLOW:ErrIncompatibleSchedulerRequest"),
	)
	ErrDispatchTaskRequestIDNotFound = errors.Normalize(
		"dispatch task request id %s not found",
		errors.RFCCodeText("DFLOW:ErrDispatchTaskRequestIDNotFound"),
	)
	ErrElectionRecordConflict = errors.Normalize(
		"election record conflict",
		errors.RFCCodeText("DFLOW:ErrElectionRecordConflict"),
	)
	ErrDeserializeConfig = errors.Normalize(
		"deserialize config failed",
		errors.RFCCodeText("DFLOW:ErrDeserializeConfig"),
	)
)

all dataflow engine errors

var (
	// Is is a shortcut for errors.Is.
	Is = errors.Is
	// As is a shortcut for errors.As.
	As = errors.As
	// New is a shortcut for github.com/pingcap/errors.New.
	New = perrors.New
	// Errorf is a shortcut for github.com/pingcap/errors.Errorf.
	Errorf = perrors.Errorf
	// Trace is a shortcut for github.com/pingcap/errors.Trace.
	Trace = perrors.Trace
	// Cause is a shortcut for github.com/pingcap/errors.Cause.
	Cause = perrors.Cause
	// Annotate is a shortcut for github.com/pingcap/errors.Annotate.
	Annotate = perrors.Annotate
	// Annotatef is a shortcut for github.com/pingcap/errors.Annotatef.
	Annotatef = perrors.Annotatef
	// WithMessage is a shortcut for github.com/pingcap/errors.WithMessage.
	WithMessage = perrors.WithMessage
)

var ChangeFeedGCFastFailError = []*errors.Error{
	ErrGCTTLExceeded, ErrSnapshotLostByGC, ErrStartTsBeforeGC,
}

ChangeFeedGCFastFailError is read only. If this type of error occurs in a changefeed, it means that the data it wants to replicate has been or will be GC. So it makes no sense to try to resume the changefeed, and the changefeed should immediately be failed.

Functions

func GRPCStatusCode

func GRPCStatusCode(err error) codes.Code

GRPCStatusCode returns the gRPC status code for the given error.

func HTTPStatusCode

func HTTPStatusCode(err error) int

HTTPStatusCode returns the HTTP status code for the given error.

func IsChangefeedGCFastFailError

func IsChangefeedGCFastFailError(err error) bool

IsChangefeedGCFastFailError checks if an error is a ChangefeedFastFailError

func IsChangefeedGCFastFailErrorCode

func IsChangefeedGCFastFailErrorCode(errCode errors.RFCErrorCode) bool

IsChangefeedGCFastFailErrorCode checks the error code, returns true if it is a ChangefeedFastFailError code

func IsCliUnprintableError

func IsCliUnprintableError(err error) bool

IsCliUnprintableError returns true if the error should not be printed in cli.

func IsContextCanceledError

func IsContextCanceledError(err error) bool

IsContextCanceledError checks if an error is caused by context.Canceled.

func IsContextDeadlineExceededError

func IsContextDeadlineExceededError(err error) bool

IsContextDeadlineExceededError checks if an error is caused by context.DeadlineExceeded.

func IsDupEntryError

func IsDupEntryError(err error) bool

IsDupEntryError checks if an error is a duplicate entry error.

func IsRetryableError

func IsRetryableError(err error) bool

IsRetryableError check the error is safe or worth to retry

func OriginError

func OriginError(err error) error

OriginError return original err

func RFCCode

func RFCCode(err error) (errors.RFCErrorCode, bool)

RFCCode returns a RFCCode from an error

func ShouldFailChangefeed

func ShouldFailChangefeed(err error) bool

ShouldFailChangefeed returns true if an error is a changefeed not retry error.

func ToDMError

func ToDMError(err error) error

ToDMError tries best to construct a DM terror from an error object - If err wraps a DM standard error, return it directly. - If not, checks whether error message matches DM standard error format

• If it does, construct a DM terror from the error code and message.
• Otherwise returns the original error

func WrapChangefeedUnretryableErr

func WrapChangefeedUnretryableErr(err error, args ...interface{}) error

WrapChangefeedUnretryableErr wraps an error into ErrChangefeedUnRetryable.

func WrapError

func WrapError(rfcError *errors.Error, err error, args ...interface{}) error

WrapError generates a new error based on given `*errors.Error`, wraps the err as cause error. If given `err` is nil, returns a nil error, which a the different behavior against `Wrap` function in pingcap/errors.