Repository: pingcap/pd
Pull request #1938 (2019-11-13 13:47:32)

operator: OpStatus & OpStatusTracker
<!--
Thank you for working on PD! Please read PD's [CONTRIBUTING](https://github.com/pingcap/pd/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve? <!--add the issue link with summary if it exists-->

The operator status is hidden in `createTime` and `startTime`, which is not easy to use.

### What is changed and how it works?

Add `OpStatus` and `OpStatusTracker` to maintain status.
There still some work need to do:

- [x] Update `OperatorController`
- [x] Unit test for `OpStatus` and `OpStatusTracker`
- [x] Unit test for new `Operator`
- [x] Unit test for new `OperatorController`

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test

codecov-io: # [Codecov](https://codecov.io/gh/pingcap/pd/pull/1938?src=pr&el=h1) Report
> Merging [#1938](https://codecov.io/gh/pingcap/pd/pull/1938?src=pr&el=desc) into [master](https://codecov.io/gh/pingcap/pd/commit/2488cb9786444100f163bd5ce53faf835c1fc42f?src=pr&el=desc) will **decrease** coverage by `0.2%`.
> The diff coverage is `77.24%`.

[![Impacted file tree graph](https://codecov.io/gh/pingcap/pd/pull/1938/graphs/tree.svg?width=650&token=J1yRL471OI&height=150&src=pr)](https://codecov.io/gh/pingcap/pd/pull/1938?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master    #1938      +/-   ##
==========================================
- Coverage   77.94%   77.73%   -0.21%     
==========================================
  Files         170      173       +3     
  Lines       16860    17057     +197     
==========================================
+ Hits        13141    13260     +119     
- Misses       2698     2763      +65     
- Partials     1021     1034      +13
```


| [Impacted Files](https://codecov.io/gh/pingcap/pd/pull/1938?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [server/schedule/test\_util.go](https://codecov.io/gh/pingcap/pd/pull/1938/diff?src=pr&el=tree#diff-c2VydmVyL3NjaGVkdWxlL3Rlc3RfdXRpbC5nbw==) | `66.12% <100%> (+0.55%)` | :arrow_up: |
| [server/schedule/operator/test\_util.go](https://codecov.io/gh/pingcap/pd/pull/1938/diff?src=pr&el=tree#diff-c2VydmVyL3NjaGVkdWxlL29wZXJhdG9yL3Rlc3RfdXRpbC5nbw==) | `50% <50%> (ø)` | |
| [server/schedule/operator\_controller.go](https://codecov.io/gh/pingcap/pd/pull/1938/diff?src=pr&el=tree#diff-c2VydmVyL3NjaGVkdWxlL29wZXJhdG9yX2NvbnRyb2xsZXIuZ28=) | `84.64% <73%> (-6.43%)` | :arrow_down: |
| [server/schedule/operator/status.go](https://codecov.io/gh/pingcap/pd/pull/1938/diff?src=pr&el=tree#diff-c2VydmVyL3NjaGVkdWxlL29wZXJhdG9yL3N0YXR1cy5nbw==) | `80% <80%> (ø)` | |
| [server/schedule/operator/status\_tracker.go](https://codecov.io/gh/pingcap/pd/pull/1938/diff?src=pr&el=tree#diff-c2VydmVyL3NjaGVkdWxlL29wZXJhdG9yL3N0YXR1c190cmFja2VyLmdv) | `81.81% <81.81%> (ø)` | |
| [server/schedule/operator/operator.go](https://codecov.io/gh/pingcap/pd/pull/1938/diff?src=pr&el=tree#diff-c2VydmVyL3NjaGVkdWxlL29wZXJhdG9yL29wZXJhdG9yLmdv) | `93.19% <86.66%> (-4.48%)` | :arrow_down: |
| [tools/pd-ctl/pdctl/command/global.go](https://codecov.io/gh/pingcap/pd/pull/1938/diff?src=pr&el=tree#diff-dG9vbHMvcGQtY3RsL3BkY3RsL2NvbW1hbmQvZ2xvYmFsLmdv) | `66.03% <0%> (-5.4%)` | :arrow_down: |
| [server/schedulers/random\_merge.go](https://codecov.io/gh/pingcap/pd/pull/1938/diff?src=pr&el=tree#diff-c2VydmVyL3NjaGVkdWxlcnMvcmFuZG9tX21lcmdlLmdv) | `61.81% <0%> (-3.64%)` | :arrow_down: |
| [server/tso/tso.go](https://codecov.io/gh/pingcap/pd/pull/1938/diff?src=pr&el=tree#diff-c2VydmVyL3Rzby90c28uZ28=) | `77.37% <0%> (-2.92%)` | :arrow_down: |
| [pkg/etcdutil/etcdutil.go](https://codecov.io/gh/pingcap/pd/pull/1938/diff?src=pr&el=tree#diff-cGtnL2V0Y2R1dGlsL2V0Y2R1dGlsLmdv) | `88.4% <0%> (-2.9%)` | :arrow_down: |
| ... and [13 more](https://codecov.io/gh/pingcap/pd/pull/1938/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pingcap/pd/pull/1938?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/pingcap/pd/pull/1938?src=pr&el=footer). Last update [2488cb9...2d966e5](https://codecov.io/gh/pingcap/pd/pull/1938?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

nolouch: PTAL @lhy1024 
nolouch: /merge
sre-bot: /run-all-tests

Patch:
diff --git a/server/schedule/operator/operator.go b/server/schedule/operator/operator.go
index 03870e25d..bbc2e108a 100644
--- a/server/schedule/operator/operator.go
+++ b/server/schedule/operator/operator.go
@@ -25,10 +25,13 @@ import (
 )
 
 const (
-	// LeaderOperatorWaitTime is the duration that when a leader operator lives
+	// OperatorExpireTime is the duration that when an operator is not started
+	// after it, the operator will be considered expired.
+	OperatorExpireTime = 3 * time.Second
+	// LeaderOperatorWaitTime is the duration that when a leader operator runs
 	// longer than it, the operator will be considered timeout.
 	LeaderOperatorWaitTime = 10 * time.Second
-	// RegionOperatorWaitTime is the duration that when a region operator lives
+	// RegionOperatorWaitTime is the duration that when a region operator runs
 	// longer than it, the operator will be considered timeout.
 	RegionOperatorWaitTime = 10 * time.Minute
 )
@@ -49,11 +52,9 @@ type Operator struct {
 	kind        OpKind
 	steps       []OpStep
 	currentStep int32
-	createTime  time.Time
-	// startTime is used to record the start time of an operator which is added into running operators.
-	startTime time.Time
-	stepTime  int64
-	level     core.PriorityLevel
+	status      OpStatusTracker
+	stepTime    int64
+	level       core.PriorityLevel
 }
 
 // NewOperator creates a new operator.
@@ -69,8 +70,7 @@ func NewOperator(desc, brief string, regionID uint64, regionEpoch *metapb.Region
 		regionEpoch: regionEpoch,
 		kind:        kind,
 		steps:       steps,
-		createTime:  time.Now(),
-		stepTime:    time.Now().UnixNano(),
+		status:      NewOpStatusTracker(),
 		level:       level,
 	}
 }
@@ -80,13 +80,13 @@ func (o *Operator) String() string {
 	for i := range o.steps {
 		stepStrs[i] = o.steps[i].String()
 	}
-	s := fmt.Sprintf("%s {%s} (kind:%s, region:%v(%v,%v), createAt:%s, startAt:%s, currentStep:%v, steps:[%s])", o.desc, o.brief, o.kind, o.regionID, o.regionEpoch.GetVersion(), o.regionEpoch.GetConfVer(), o.createTime, o.startTime, atomic.LoadInt32(&o.currentStep), strings.Join(stepStrs, ", "))
-	if o.IsTimeout() {
-		s = s + " timeout"
-	}
-	if o.IsFinish() {
+	s := fmt.Sprintf("%s {%s} (kind:%s, region:%v(%v,%v), createAt:%s, startAt:%s, currentStep:%v, steps:[%s])", o.desc, o.brief, o.kind, o.regionID, o.regionEpoch.GetVersion(), o.regionEpoch.GetConfVer(), o.GetCreateTime(), o.GetStartTime(), atomic.LoadInt32(&o.currentStep), strings.Join(stepStrs, ", "))
+	if o.CheckSuccess() {
 		s = s + " finished"
 	}
+	if o.CheckTimeout() {
+		s = s + " timeout"
+	}
 	return s
 }
 
@@ -125,24 +125,90 @@ func (o *Operator) Kind() OpKind {
 	return o.kind
 }
 
+// Status returns operator status.
+func (o *Operator) Status() OpStatus {
+	return o.status.Status()
+}
+
+// GetReachTimeOf returns the time when operator reaches the given status.
+func (o *Operator) GetReachTimeOf(st OpStatus) time.Time {
+	return o.status.ReachTimeOf(st)
+}
+
+// GetCreateTime gets the create time of operator.
+func (o *Operator) GetCreateTime() time.Time {
+	return o.status.ReachTimeOf(CREATED)
+}
+
 // ElapsedTime returns duration since it was created.
 func (o *Operator) ElapsedTime() time.Duration {
-	return time.Since(o.createTime)
+	return time.Since(o.GetCreateTime())
 }
 
-// RunningTime returns duration since it was promoted.
-func (o *Operator) RunningTime() time.Duration {
-	return time.Since(o.startTime)
+// Start sets the operator to STARTED status, returns whether succeeded.
+func (o *Operator) Start() bool {
+	if o.status.To(STARTED) {
+		atomic.StoreInt64(&o.stepTime, time.Now().UnixNano())
+		return true
+	}
+	return false
 }
 
-// SetStartTime sets the start time for operator.
-func (o *Operator) SetStartTime(t time.Time) {
-	o.startTime = t
+// HasStarted returns whether operator has started.
+func (o *Operator) HasStarted() bool {
+	return !o.GetStartTime().IsZero()
 }
 
-// GetStartTime ges the start time for operator.
+// GetStartTime gets the start time of operator.
 func (o *Operator) GetStartTime() time.Time {
-	return o.startTime
+	return o.status.ReachTimeOf(STARTED)
+}
+
+// RunningTime returns duration since it started.
+func (o *Operator) RunningTime() time.Duration {
+	if o.HasStarted() {
+		return time.Since(o.GetStartTime())
+	}
+	return 0
+}
+
+// IsEnd checks if the operator is at and end status.
+func (o *Operator) IsEnd() bool {
+	return o.status.IsEnd()
+}
+
+// CheckSuccess checks if all steps are finished, and update the status.
+func (o *Operator) CheckSuccess() bool {
+	if atomic.LoadInt32(&o.currentStep) >= int32(len(o.steps)) {
+		return o.status.To(SUCCESS) || o.Status() == SUCCESS
+	}
+	return false
+}
+
+// Cancel marks the operator canceled.
+func (o *Operator) Cancel() bool {
+	return o.status.To(CANCELED)
+}
+
+// Replace marks the operator replaced.
+func (o *Operator) Replace() bool {
+	return o.status.To(REPLACED)
+}
+
+// CheckExpired checks if the operator is expired, and update the status.
+func (o *Operator) CheckExpired() bool {
+	return o.status.CheckExpired(OperatorExpireTime)
+}
+
+// CheckTimeout checks if the operator is timeout, and update the status.
+func (o *Operator) CheckTimeout() bool {
+	if o.CheckSuccess() {
+		return false
+	}
+	if o.kind&OpRegion != 0 {
+		return o.status.CheckTimeout(RegionOperatorWaitTime)
+	}
+	return o.status.CheckTimeout(LeaderOperatorWaitTime)
 }
 
 // Len returns the operator's steps count.
@@ -159,8 +225,15 @@ func (o *Operator) Step(i int) OpStep {
 }
 
 // Check checks if current step is finished, returns next step to take action.
+// If operator is at an end status, check returns nil.
 // It's safe to be called by multiple goroutine concurrently.
+// FIXME: metrics is not thread-safe
 func (o *Operator) Check(region *core.RegionInfo) OpStep {
+	if o.IsEnd() {
+		return nil
+	}
+	// CheckTimeout will call CheckSuccess first
+	defer func() { _ = o.CheckTimeout() }()
 	for step := atomic.LoadInt32(&o.currentStep); int(step) < len(o.steps); step++ {
 		if o.steps[int(step)].IsFinish(region) {
 			operatorStepDuration.WithLabelValues(reflect.TypeOf(o.steps[int(step)]).Name()).
@@ -200,31 +273,6 @@ func (o *Operator) GetPriorityLevel() core.PriorityLevel {
 	return o.level
 }
 
-// IsFinish checks if all steps are finished.
-func (o *Operator) IsFinish() bool {
-	return atomic.LoadInt32(&o.currentStep) >= int32(len(o.steps))
-}
-
-// IsTimeout checks the operator's create time and determines if it is timeout.
-func (o *Operator) IsTimeout() bool {
-	var timeout bool
-	if o.IsFinish() {
-		return false
-	}
-	if o.startTime.IsZero() {
-		return false
-	}
-	if o.kind&OpRegion != 0 {
-		timeout = time.Since(o.startTime) > RegionOperatorWaitTime
-	} else {
-		timeout = time.Since(o.startTime) > LeaderOperatorWaitTime
-	}
-	if timeout {
-		return true
-	}
-	return false
-}
-
 // UnfinishedInfluence calculates the store difference which unfinished operator steps make.
 func (o *Operator) UnfinishedInfluence(opInfluence OpInfluence, region *core.RegionInfo) {
 	for step := atomic.LoadInt32(&o.currentStep); int(step) < len(o.steps); step++ {
diff --git a/server/schedule/operator/status.go b/server/schedule/operator/status.go
new file mode 100644
index 000000000..75b962cd4
--- /dev/null
+++ b/server/schedule/operator/status.go
@@ -0,0 +1,106 @@
+// Copyright 2019 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package operator
+
+import (
+	"github.com/pingcap/kvproto/pkg/pdpb"
+)
+
+// OpStatus represents the status of an Operator.
+type OpStatus = uint32
+
+// Status list
+const (
+	// Status list
+	// Just created. Next status: {RUNNING, CANCELED, EXPIRED}.
+	CREATED OpStatus = iota
+	// Started and not finished. Next status: {SUCCESS, CANCELED, REPLACED, TIMEOUT}.
+	STARTED
+	// Followings are end status, i.e. no next status.
+	SUCCESS  // Finished successfully
+	CANCELED // Canceled due to some reason
+	REPLACED // Replaced by an higher priority operator
+	EXPIRED  // Didn't start to run for too long
+	TIMEOUT  // Running for too long
+	// Status list end
+	statusCount    // Total count of status
+	firstEndStatus = SUCCESS
+)
+
+type transition [statusCount][statusCount]bool
+
+// Valid status transition
+var validTrans transition = transition{
+	CREATED: {
+		STARTED:  true,
+		CANCELED: true,
+		EXPIRED:  true,
+	},
+	STARTED: {
+		SUCCESS:  true,
+		CANCELED: true,
+		REPLACED: true,
+		TIMEOUT:  true,
+	},
+	SUCCESS:  {},
+	CANCELED: {},
+	REPLACED: {},
+	EXPIRED:  {},
+	TIMEOUT:  {},
+}
+
+var statusString [statusCount]string = [statusCount]string{
+	CREATED:  "Created",
+	STARTED:  "Started",
+	SUCCESS:  "Success",
+	CANCELED: "Canceled",
+	REPLACED: "Replaced",
+	EXPIRED:  "Expired",
+	TIMEOUT:  "Timeout",
+}
+
+const invalid pdpb.OperatorStatus = pdpb.OperatorStatus_RUNNING + 1
+
+var pdpbStatus [statusCount]pdpb.OperatorStatus = [statusCount]pdpb.OperatorStatus{
+	// FIXME: use a valid status
+	CREATED:  invalid,
+	STARTED:  pdpb.OperatorStatus_RUNNING,
+	SUCCESS:  pdpb.OperatorStatus_SUCCESS,
+	CANCELED: pdpb.OperatorStatus_CANCEL,
+	REPLACED: pdpb.OperatorStatus_REPLACE,
+	// FIXME: use a better status
+	EXPIRED: pdpb.OperatorStatus_TIMEOUT,
+	TIMEOUT: pdpb.OperatorStatus_TIMEOUT,
+}
+
+// IsEndStatus checks whether s is an end status.
+func IsEndStatus(s OpStatus) bool {
+	return firstEndStatus <= s && s < statusCount
+}
+
+// OpStatusToPDPB converts OpStatus to pdpb.OperatorStatus.
+func OpStatusToPDPB(s OpStatus) pdpb.OperatorStatus {
+	if s < statusCount {
+		return pdpbStatus[s]
+	}
+	return invalid
+}
+
+// OpStatusToString converts Status to string.
+func OpStatusToString(s OpStatus) string {
+	if s < statusCount {
+		return statusString[s]
+	}
+	return "Unknown"
+}
diff --git a/server/schedule/operator/status_tracker.go b/server/schedule/operator/status_tracker.go
new file mode 100644
index 000000000..88d980851
--- /dev/null
+++ b/server/schedule/operator/status_tracker.go
@@ -0,0 +1,137 @@
+// Copyright 2019 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package operator
+
+import (
+	"sync"
+	"time"
+)
+
+// Only record non-end status and one end status.
+type statusTimes [firstEndStatus + 1]time.Time
+
+// OpStatusTracker represents the status of an operator.
+type OpStatusTracker struct {
+	rw         sync.RWMutex
+	current    OpStatus    // Current status
+	reachTimes statusTimes // Time when reach the current status
+}
+
+// NewOpStatusTracker creates an OpStatus.
+func NewOpStatusTracker() OpStatusTracker {
+	return OpStatusTracker{
+		current:    CREATED,
+		reachTimes: statusTimes{CREATED: time.Now()},
+	}
+}
+
+// Status returns current status.
+func (trk *OpStatusTracker) Status() OpStatus {
+	trk.rw.RLock()
+	defer trk.rw.RUnlock()
+	return trk.current
+}
+
+// ReachTime returns the reach time of current status.
+func (trk *OpStatusTracker) ReachTime() time.Time {
+	trk.rw.RLock()
+	defer trk.rw.RUnlock()
+	return trk.getTime(trk.current)
+}
+
+// ReachTimeOf returns the time when reached given status. If didn't reatched the given status, return zero.
+func (trk *OpStatusTracker) ReachTimeOf(s OpStatus) time.Time {
+	trk.rw.RLock()
+	defer trk.rw.RUnlock()
+	return trk.getTime(s)
+}
+
+func (trk *OpStatusTracker) getTime(s OpStatus) time.Time {
+	if s < firstEndStatus {
+		return trk.reachTimes[s]
+	} else if trk.current == s {
+		return trk.reachTimes[firstEndStatus]
+	} else {
+		return time.Time{}
+	}
+}
+
+// To transfer the current status to dst if this transition is valid,
+// returns whether transferred.
+func (trk *OpStatusTracker) To(dst OpStatus) bool {
+	trk.rw.Lock()
+	defer trk.rw.Unlock()
+	return trk.toLocked(dst)
+}
+
+func (trk *OpStatusTracker) toLocked(dst OpStatus) bool {
+	if dst < statusCount && validTrans[trk.current][dst] {
+		trk.current = dst
+		trk.setTime(trk.current, time.Now())
+		return true
+	}
+	return false
+}
+
+func (trk *OpStatusTracker) setTime(st OpStatus, t time.Time) {
+	if st < firstEndStatus {
+		trk.reachTimes[st] = t
+	} else {
+		trk.reachTimes[firstEndStatus] = t
+	}
+}
+
+// IsEnd checks whether the current status is an end status.
+func (trk *OpStatusTracker) IsEnd() bool {
+	trk.rw.RLock()
+	defer trk.rw.RUnlock()
+	return IsEndStatus(trk.current)
+}
+
+// CheckExpired checks if expired, and update the current status.
+func (trk *OpStatusTracker) CheckExpired(exp time.Duration) bool {
+	trk.rw.Lock()
+	defer trk.rw.Unlock()
+	switch trk.current {
+	case CREATED:
+		if time.Since(trk.reachTimes[CREATED]) < exp {
+			return false
+		}
+		_ = trk.toLocked(EXPIRED)
+		return true
+	}
+	return trk.current == EXPIRED
+}
+
+// CheckTimeout checks if timeout, and update the current status.
+func (trk *OpStatusTracker) CheckTimeout(wait time.Duration) bool {
+	trk.rw.Lock()
+	defer trk.rw.Unlock()
+	switch trk.current {
+	case STARTED:
+		if time.Since(trk.reachTimes[STARTED]) < wait {
+			return false
+		}
+		_ = trk.toLocked(TIMEOUT)
+		return true
+	}
+	return trk.current == TIMEOUT
+}
+
+// String implements fmt.Stringer.
+func (trk *OpStatusTracker) String() string {
+	trk.rw.RLock()
+	defer trk.rw.RUnlock()
+	return OpStatusToString(trk.current)
+}
diff --git a/server/schedule/operator_controller.go b/server/schedule/operator_controller.go
index 3a3ab1b6a..a573798fe 100644
--- a/server/schedule/operator_controller.go
+++ b/server/schedule/operator_controller.go
@@ -110,53 +110,74 @@ func (oc *OperatorController) Dispatch(region *core.RegionInfo, source string) {
 		failpoint.Inject("concurrentRemoveOperator", func() {
 			time.Sleep(500 * time.Millisecond)
 		})
-		timeout := op.IsTimeout()
-		if step := op.Check(region); step != nil && !timeout {
-			operatorCounter.WithLabelValues(op.Desc(), "check").Inc()
 
-			// When the "source" is heartbeat, the region may have a newer
-			// confver than the region that the operator holds. In this case,
-			// the operator is stale, and will not be executed even we would
-			// have sent it to TiKV servers. Here, we just cancel it.
-			origin := op.RegionEpoch()
-			latest := region.GetRegionEpoch()
-			changes := latest.GetConfVer() - origin.GetConfVer()
-			if source == DispatchFromHeartBeat &&
-				changes > uint64(op.ConfVerChanged(region)) {
-
-				if oc.RemoveOperator(op) {
-					log.Info("stale operator",
-						zap.Uint64("region-id", region.GetID()),
-						zap.Duration("takes", op.RunningTime()),
-						zap.Reflect("operator", op),
-						zap.Reflect("latest-epoch", region.GetRegionEpoch()),
-						zap.Uint64("diff", changes),
-					)
-					operatorCounter.WithLabelValues(op.Desc(), "stale").Inc()
-					oc.opRecords.Put(op, pdpb.OperatorStatus_CANCEL)
-					oc.PromoteWaitingOperator()
-				}
+		// Update operator status:
+		// The operator status should be STARTED.
+		// Check will call CheckSuccess and CheckTimeout.
+		step := op.Check(region)
 
+		switch op.Status() {
+		case operator.STARTED:
+			operatorCounter.WithLabelValues(op.Desc(), "check").Inc()
+			if source == DispatchFromHeartBeat && oc.checkStaleOperator(op, region) {
 				return
 			}
-
 			oc.SendScheduleCommand(region, step, source)
-			return
-		}
-		if op.IsFinish() && oc.RemoveOperator(op) {
-			log.Info("operator finish", zap.Uint64("region-id", region.GetID()), zap.Duration("takes", op.RunningTime()), zap.Reflect("operator", op))
-			operatorCounter.WithLabelValues(op.Desc(), "finish").Inc()
-			operatorDuration.WithLabelValues(op.Desc()).Observe(op.RunningTime().Seconds())
+		case operator.SUCCESS:
 			oc.pushHistory(op)
-			oc.opRecords.Put(op, pdpb.OperatorStatus_SUCCESS)
-			oc.PromoteWaitingOperator()
-		} else if timeout && oc.RemoveOperator(op) {
-			log.Info("operator timeout", zap.Uint64("region-id", region.GetID()), zap.Duration("takes", op.RunningTime()), zap.Reflect("operator", op))
-			operatorCounter.WithLabelValues(op.Desc(), "timeout").Inc()
-			oc.opRecords.Put(op, pdpb.OperatorStatus_TIMEOUT)
+			if oc.RemoveOperator(op) {
+				oc.PromoteWaitingOperator()
+			}
+		case operator.TIMEOUT:
+			if oc.RemoveOperator(op) {
+				oc.PromoteWaitingOperator()
+			}
+		default:
+			if oc.removeOperatorWithoutBury(op) {
+				// CREATED, EXPIRED must not appear.
+				// CANCELED, REPLACED must remove before transition.
+				log.Error("dispatching operator with unexpected status",
+					zap.Uint64("region-id", op.RegionID()),
+					zap.String("status", operator.OpStatusToString(op.Status())),
+					zap.Reflect("operator", op))
+				operatorCounter.WithLabelValues(op.Desc(), "unexpected").Inc()
+				failpoint.Inject("unexpectedOperator", func() {
+					panic(op)
+				})
+				_ = op.Cancel()
+				oc.buryOperator(op)
+				oc.PromoteWaitingOperator()
+			}
+		}
+	}
+}
+
+func (oc *OperatorController) checkStaleOperator(op *operator.Operator, region *core.RegionInfo) bool {
+	// When the "source" is heartbeat, the region may have a newer
+	// confver than the region that the operator holds. In this case,
+	// the operator is stale, and will not be executed even we would
+	// have sent it to TiKV servers. Here, we just cancel it.
+	origin := op.RegionEpoch()
+	latest := region.GetRegionEpoch()
+	changes := latest.GetConfVer() - origin.GetConfVer()
+	if changes > uint64(op.ConfVerChanged(region)) {
+
+		if oc.removeOperatorWithoutBury(op) {
+			if op.Cancel() {
+				log.Info("stale operator",
+					zap.Uint64("region-id", op.RegionID()),
+					zap.Duration("takes", op.RunningTime()),
+					zap.Reflect("operator", op),
+					zap.Reflect("latest-epoch", region.GetRegionEpoch()),
+					zap.Uint64("diff", changes),
+				)
+				operatorCounter.WithLabelValues(op.Desc(), "stale").Inc()
+			}
 			oc.PromoteWaitingOperator()
 		}
+		return true
 	}
+	return false
 }
 
 func (oc *OperatorController) getNextPushOperatorTime(step operator.OpStep, now time.Time) time.Time {
@@ -186,11 +207,13 @@ func (oc *OperatorController) pollNeedDispatchRegion() (r *core.RegionInfo, next
 	r = oc.cluster.GetRegion(regionID)
 	if r == nil {
 		_ = oc.removeOperatorLocked(op)
-		log.Debug("remove operator because region disappeared",
-			zap.Uint64("region-id", op.RegionID()),
-			zap.Stringer("operator", op))
-		operatorCounter.WithLabelValues(op.Desc(), "disappear").Inc()
-		oc.opRecords.Put(op, pdpb.OperatorStatus_CANCEL)
+		if op.Cancel() {
+			log.Warn("remove operator because region disappeared",
+				zap.Uint64("region-id", op.RegionID()),
+				zap.Stringer("operator", op))
+			operatorCounter.WithLabelValues(op.Desc(), "disappear").Inc()
+		}
+		oc.buryOperator(op)
 		return nil, true
 	}
 	step := op.Check(r)
@@ -231,7 +254,8 @@ func (oc *OperatorController) AddWaitingOperator(ops ...*operator.Operator) bool
 	if !oc.checkAddOperator(ops...) {
 		for _, op := range ops {
 			operatorWaitCounter.WithLabelValues(op.Desc(), "add_canceled").Inc()
-			oc.opRecords.Put(op, pdpb.OperatorStatus_CANCEL)
+			_ = op.Cancel()
+			oc.buryOperator(op)
 		}
 		oc.Unlock()
 		return false
@@ -241,6 +265,10 @@ func (oc *OperatorController) AddWaitingOperator(ops ...*operator.Operator) bool
 	desc := op.Desc()
 	if oc.wopStatus.ops[desc] >= oc.cluster.GetSchedulerMaxWaitingOperator() {
 		operatorWaitCounter.WithLabelValues(op.Desc(), "exceed_max").Inc()
+		for _, op := range ops {
+			_ = op.Cancel()
+			oc.buryOperator(op)
+		}
 		oc.Unlock()
 		return false
 	}
@@ -264,12 +292,15 @@ func (oc *OperatorController) AddOperator(ops ...*operator.Operator) bool {
 	if oc.exceedStoreLimit(ops...) || !oc.checkAddOperator(ops...) {
 		for _, op := range ops {
 			operatorCounter.WithLabelValues(op.Desc(), "cancel").Inc()
-			oc.opRecords.Put(op, pdpb.OperatorStatus_CANCEL)
+			_ = op.Cancel()
+			oc.buryOperator(op)
 		}
 		return false
 	}
 	for _, op := range ops {
-		oc.addOperatorLocked(op)
+		if !oc.addOperatorLocked(op) {
+			return false
+		}
 	}
 	return true
 }
@@ -289,7 +320,8 @@ func (oc *OperatorController) PromoteWaitingOperator() {
 		if oc.exceedStoreLimit(ops...) || !oc.checkAddOperator(ops...) {
 			for _, op := range ops {
 				operatorWaitCounter.WithLabelValues(op.Desc(), "promote_canceled").Inc()
-				oc.opRecords.Put(op, pdpb.OperatorStatus_CANCEL)
+				_ = op.Cancel()
+				oc.buryOperator(op)
 			}
 			oc.wopStatus.ops[ops[0].Desc()]--
 			continue
@@ -299,7 +331,9 @@ func (oc *OperatorController) PromoteWaitingOperator() {
 	}
 
 	for _, op := range ops {
-		oc.addOperatorLocked(op)
+		if !oc.addOperatorLocked(op) {
+			break
+		}
 	}
 }
 
@@ -308,45 +342,80 @@ func (oc *OperatorController) PromoteWaitingOperator() {
 // - There is no such region in the cluster
 // - The epoch of the operator and the epoch of the corresponding region are no longer consistent.
 // - The region already has a higher priority or same priority operator.
+// - At least one operator is expired.
 func (oc *OperatorController) checkAddOperator(ops ...*operator.Operator) bool {
 	for _, op := range ops {
 		region := oc.cluster.GetRegion(op.RegionID())
 		if region == nil {
-			log.Debug("region not found, cancel add operator", zap.Uint64("region-id", op.RegionID()))
+			log.Debug("region not found, cancel add operator",
+				zap.Uint64("region-id", op.RegionID()))
 			return false
 		}
-		if region.GetRegionEpoch().GetVersion() != op.RegionEpoch().GetVersion() || region.GetRegionEpoch().GetConfVer() != op.RegionEpoch().GetConfVer() {
-			log.Debug("region epoch not match, cancel add operator", zap.Uint64("region-id", op.RegionID()), zap.Reflect("old", region.GetRegionEpoch()), zap.Reflect("new", op.RegionEpoch()))
+		if region.GetRegionEpoch().GetVersion() != op.RegionEpoch().GetVersion() ||
+			region.GetRegionEpoch().GetConfVer() != op.RegionEpoch().GetConfVer() {
+			log.Debug("region epoch not match, cancel add operator",
+				zap.Uint64("region-id", op.RegionID()),
+				zap.Reflect("old", region.GetRegionEpoch()),
+				zap.Reflect("new", op.RegionEpoch()))
 			return false
 		}
 		if old := oc.operators[op.RegionID()]; old != nil && !isHigherPriorityOperator(op, old) {
-			log.Debug("already have operator, cancel add operator", zap.Uint64("region-id", op.RegionID()), zap.Reflect("old", old))
+			log.Debug("already have operator, cancel add operator",
+				zap.Uint64("region-id", op.RegionID()),
+				zap.Reflect("old", old))
+			return false
+		}
+		if op.Status() != operator.CREATED {
+			log.Error("trying to add operator with unexpected status",
+				zap.Uint64("region-id", op.RegionID()),
+				zap.String("status", operator.OpStatusToString(op.Status())),
+				zap.Reflect("operator", op))
+			failpoint.Inject("unexpectedOperator", func() {
+				panic(op)
+			})
 			return false
 		}
 	}
-	return true
+	expired := false
+	for _, op := range ops {
+		if op.CheckExpired() {
+			expired = true
+		}
+	}
+	return !expired
 }
 
 func isHigherPriorityOperator(new, old *operator.Operator) bool {
 	return new.GetPriorityLevel() > old.GetPriorityLevel()
 }
 
-func (oc *OperatorController) addOperatorLocked(op *operator.Operator) {
+func (oc *OperatorController) addOperatorLocked(op *operator.Operator) bool {
 	regionID := op.RegionID()
 
-	log.Info("add operator", zap.Uint64("region-id", regionID), zap.Reflect("operator", op))
+	log.Info("add operator",
+		zap.Uint64("region-id", regionID),
+		zap.Reflect("operator", op))
 
 	// If there is an old operator, replace it. The priority should be checked
 	// already.
 	if old, ok := oc.operators[regionID]; ok {
 		_ = oc.removeOperatorLocked(old)
-		log.Info("replace old operator", zap.Uint64("region-id", regionID), zap.Duration("takes", old.RunningTime()), zap.Reflect("operator", old))
-		operatorCounter.WithLabelValues(old.Desc(), "replace").Inc()
-		oc.opRecords.Put(old, pdpb.OperatorStatus_REPLACE)
+		_ = old.Replace()
+		oc.buryOperator(old)
 	}
 
+	if !op.Start() {
+		log.Error("adding operator with unexpected status",
+			zap.Uint64("region-id", regionID),
+			zap.String("status", operator.OpStatusToString(op.Status())),
+			zap.Reflect("operator", op))
+		failpoint.Inject("unexpectedOperator", func() {
+			panic(op)
+		})
+		operatorCounter.WithLabelValues(op.Desc(), "unexpected").Inc()
+		return false
+	}
 	oc.operators[regionID] = op
-	op.SetStartTime(time.Now())
 	operatorCounter.WithLabelValues(op.Desc(), "start").Inc()
 	operatorWaitDuration.WithLabelValues(op.Desc()).Observe(op.ElapsedTime().Seconds())
 	opInfluence := NewTotalOpInfluence([]*operator.Operator{op}, oc.cluster)
@@ -369,26 +438,30 @@ func (oc *OperatorController) addOperatorLocked(op *operator.Operator) {
 
 	heap.Push(&oc.opNotifierQueue, &operatorWithTime{op: op, time: oc.getNextPushOperatorTime(step, time.Now())})
 	operatorCounter.WithLabelValues(op.Desc(), "create").Inc()
+	return true
 }
 
 // RemoveOperator removes a operator from the running operators.
-func (oc *OperatorController) RemoveOperator(op *operator.Operator) (found bool) {
+func (oc *OperatorController) RemoveOperator(op *operator.Operator) bool {
 	oc.Lock()
-	defer oc.Unlock()
-	return oc.removeOperatorLocked(op)
+	removed := oc.removeOperatorLocked(op)
+	oc.Unlock()
+	if removed {
+		if op.Cancel() {
+			log.Info("operator removed",
+				zap.Uint64("region-id", op.RegionID()),
+				zap.Duration("takes", op.RunningTime()),
+				zap.Reflect("operator", op))
+		}
+		oc.buryOperator(op)
+	}
+	return removed
 }
 
-// GetOperatorStatus gets the operator and its status with the specify id.
-func (oc *OperatorController) GetOperatorStatus(id uint64) *OperatorWithStatus {
+func (oc *OperatorController) removeOperatorWithoutBury(op *operator.Operator) bool {
 	oc.Lock()
 	defer oc.Unlock()
-	if op, ok := oc.operators[id]; ok {
-		return &OperatorWithStatus{
-			Op:     op,
-			Status: pdpb.OperatorStatus_RUNNING,
-		}
-	}
-	return oc.opRecords.Get(id)
+	return oc.removeOperatorLocked(op)
 }
 
 func (oc *OperatorController) removeOperatorLocked(op *operator.Operator) bool {
@@ -402,6 +475,62 @@ func (oc *OperatorController) removeOperatorLocked(op *operator.Operator) bool {
 	return false
 }
 
+func (oc *OperatorController) buryOperator(op *operator.Operator) {
+	st := op.Status()
+
+	if !operator.IsEndStatus(st) {
+		log.Error("burying operator with non-end status",
+			zap.Uint64("region-id", op.RegionID()),
+			zap.String("status", operator.OpStatusToString(op.Status())),
+			zap.Reflect("operator", op))
+		failpoint.Inject("unexpectedOperator", func() {
+			panic(op)
+		})
+		operatorCounter.WithLabelValues(op.Desc(), "unexpected").Inc()
+		_ = op.Cancel()
+	}
+
+	switch st {
+	case operator.SUCCESS:
+		log.Info("operator finish",
+			zap.Uint64("region-id", op.RegionID()),
+			zap.Duration("takes", op.RunningTime()),
+			zap.Reflect("operator", op))
+		operatorCounter.WithLabelValues(op.Desc(), "finish").Inc()
+		operatorDuration.WithLabelValues(op.Desc()).Observe(op.RunningTime().Seconds())
+	case operator.REPLACED:
+		log.Info("replace old operator",
+			zap.Uint64("region-id", op.RegionID()),
+			zap.Duration("takes", op.RunningTime()),
+			zap.Reflect("operator", op))
+		operatorCounter.WithLabelValues(op.Desc(), "replace").Inc()
+	case operator.EXPIRED:
+		log.Info("operator expired",
+			zap.Uint64("region-id", op.RegionID()),
+			zap.Duration("lives", op.ElapsedTime()),
+			zap.Reflect("operator", op))
+		operatorCounter.WithLabelValues(op.Desc(), "expire").Inc()
+	case operator.TIMEOUT:
+		log.Info("operator timeout",
+			zap.Uint64("region-id", op.RegionID()),
+			zap.Duration("takes", op.RunningTime()),
+			zap.Reflect("operator", op))
+		operatorCounter.WithLabelValues(op.Desc(), "timeout").Inc()
+	}
+
+	oc.opRecords.Put(op)
+}
+
+// GetOperatorStatus gets the operator and its status with the specify id.
+func (oc *OperatorController) GetOperatorStatus(id uint64) *OperatorWithStatus {
+	oc.Lock()
+	defer oc.Unlock()
+	if op, ok := oc.operators[id]; ok {
+		return NewOperatorWithStatus(op)
+	}
+	return oc.opRecords.Get(id)
+}
+
 // GetOperator gets a operator from the given region.
 func (oc *OperatorController) GetOperator(regionID uint64) *operator.Operator {
 	oc.RLock()
@@ -431,7 +560,10 @@ func (oc *OperatorController) GetWaitingOperators() []*operator.Operator {
 
 // SendScheduleCommand sends a command to the region.
 func (oc *OperatorController) SendScheduleCommand(region *core.RegionInfo, step operator.OpStep, source string) {
-	log.Info("send schedule command", zap.Uint64("region-id", region.GetID()), zap.Stringer("step", step), zap.String("source", source))
+	log.Info("send schedule command",
+		zap.Uint64("region-id", region.GetID()),
+		zap.Stringer("step", step),
+		zap.String("source", source))
 	switch st := step.(type) {
 	case operator.TransferLeader:
 		cmd := &pdpb.RegionHeartbeatResponse{
@@ -611,7 +743,7 @@ func (oc *OperatorController) GetOpInfluence(cluster opt.Cluster) operator.OpInf
 	oc.RLock()
 	defer oc.RUnlock()
 	for _, op := range oc.operators {
-		if !op.IsTimeout() && !op.IsFinish() {
+		if !op.CheckTimeout() && !op.CheckSuccess() {
 			region := cluster.GetRegion(op.RegionID())
 			if region != nil {
 				op.UnfinishedInfluence(influence, region)
@@ -650,6 +782,14 @@ type OperatorWithStatus struct {
 	Status pdpb.OperatorStatus
 }
 
+// NewOperatorWithStatus creates an OperatorStatus from an operator.
+func NewOperatorWithStatus(op *operator.Operator) *OperatorWithStatus {
+	return &OperatorWithStatus{
+		Op:     op,
+		Status: operator.OpStatusToPDPB(op.Status()),
+	}
+}
+
 // MarshalJSON returns the status of operator as a JSON string
 func (o *OperatorWithStatus) MarshalJSON() ([]byte, error) {
 	return []byte(`"` + fmt.Sprintf("status: %s, operator: %s", o.Status.String(), o.Op.String()) + `"`), nil
@@ -679,12 +819,9 @@ func (o *OperatorRecords) Get(id uint64) *OperatorWithStatus {
 }
 
 // Put puts the operator and its status.
-func (o *OperatorRecords) Put(op *operator.Operator, status pdpb.OperatorStatus) {
+func (o *OperatorRecords) Put(op *operator.Operator) {
 	id := op.RegionID()
-	record := &OperatorWithStatus{
-		Op:     op,
-		Status: status,
-	}
+	record := NewOperatorWithStatus(op)
 	o.ttl.Put(id, record)
 }
 


Test patch:
diff --git a/server/api/operator_test.go b/server/api/operator_test.go
index a3f9097b6..bd4d0be0c 100644
--- a/server/api/operator_test.go
+++ b/server/api/operator_test.go
@@ -20,6 +20,7 @@ import (
 	"strings"
 
 	. "github.com/pingcap/check"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/kvproto/pkg/pdpb"
 	"github.com/pingcap/pd/server"
@@ -36,6 +37,7 @@ type testOperatorSuite struct {
 }
 
 func (s *testOperatorSuite) SetUpSuite(c *C) {
+	c.Assert(failpoint.Enable("github.com/pingcap/pd/server/schedule/unexpectedOperator", "return(true)"), IsNil)
 	s.svr, s.cleanup = mustNewServer(c, func(cfg *config.Config) { cfg.Replication.MaxReplicas = 1 })
 	mustWaitLeader(c, []*server.Server{s.svr})
 
diff --git a/server/coordinator_test.go b/server/coordinator_test.go
index 129bc4a03..37d0191dd 100644
--- a/server/coordinator_test.go
+++ b/server/coordinator_test.go
@@ -23,6 +23,7 @@ import (
 	"time"
 
 	. "github.com/pingcap/check"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/kvproto/pkg/eraftpb"
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/kvproto/pkg/pdpb"
@@ -172,6 +173,7 @@ type testCoordinatorSuite struct {
 
 func (s *testCoordinatorSuite) SetUpSuite(c *C) {
 	s.ctx, s.cancel = context.WithCancel(context.Background())
+	c.Assert(failpoint.Enable("github.com/pingcap/pd/server/schedule/unexpectedOperator", "return(true)"), IsNil)
 }
 
 func (s *testCoordinatorSuite) TearDownSuite(c *C) {
@@ -197,9 +199,10 @@ func (s *testCoordinatorSuite) TestBasic(c *C) {
 
 	// Remove the operator manually, then we can add a new operator.
 	c.Assert(oc.RemoveOperator(op1), IsTrue)
-	oc.AddWaitingOperator(op2)
-	c.Assert(oc.OperatorCount(op2.Kind()), Equals, uint64(1))
-	c.Assert(oc.GetOperator(1).RegionID(), Equals, op2.RegionID())
+	op3 := newTestOperator(1, tc.GetRegion(1).GetRegionEpoch(), operator.OpRegion)
+	oc.AddWaitingOperator(op3)
+	c.Assert(oc.OperatorCount(op3.Kind()), Equals, uint64(1))
+	c.Assert(oc.GetOperator(1).RegionID(), Equals, op3.RegionID())
 }
 
 func (s *testCoordinatorSuite) TestDispatch(c *C) {
@@ -888,6 +891,7 @@ type testOperatorControllerSuite struct {
 
 func (s *testOperatorControllerSuite) SetUpSuite(c *C) {
 	s.ctx, s.cancel = context.WithCancel(context.Background())
+	c.Assert(failpoint.Enable("github.com/pingcap/pd/server/schedule/unexpectedOperator", "return(true)"), IsNil)
 }
 
 func (s *testOperatorControllerSuite) TearDownSuite(c *C) {
@@ -903,24 +907,28 @@ func (s *testOperatorControllerSuite) TestOperatorCount(c *C) {
 
 	c.Assert(tc.addLeaderRegion(1, 1), IsNil)
 	c.Assert(tc.addLeaderRegion(2, 2), IsNil)
-	op1 := newTestOperator(1, tc.GetRegion(1).GetRegionEpoch(), operator.OpLeader)
-	oc.AddWaitingOperator(op1)
-	c.Assert(oc.OperatorCount(operator.OpLeader), Equals, uint64(1)) // 1:leader
-	op2 := newTestOperator(2, tc.GetRegion(2).GetRegionEpoch(), operator.OpLeader)
-	oc.AddWaitingOperator(op2)
-	c.Assert(oc.OperatorCount(operator.OpLeader), Equals, uint64(2)) // 1:leader, 2:leader
-	c.Assert(oc.RemoveOperator(op1), IsTrue)
-	c.Assert(oc.OperatorCount(operator.OpLeader), Equals, uint64(1)) // 2:leader
+	{
+		op1 := newTestOperator(1, tc.GetRegion(1).GetRegionEpoch(), operator.OpLeader)
+		oc.AddWaitingOperator(op1)
+		c.Assert(oc.OperatorCount(operator.OpLeader), Equals, uint64(1)) // 1:leader
+		op2 := newTestOperator(2, tc.GetRegion(2).GetRegionEpoch(), operator.OpLeader)
+		oc.AddWaitingOperator(op2)
+		c.Assert(oc.OperatorCount(operator.OpLeader), Equals, uint64(2)) // 1:leader, 2:leader
+		c.Assert(oc.RemoveOperator(op1), IsTrue)
+		c.Assert(oc.OperatorCount(operator.OpLeader), Equals, uint64(1)) // 2:leader
+	}
 
-	op1 = newTestOperator(1, tc.GetRegion(1).GetRegionEpoch(), operator.OpRegion)
-	oc.AddWaitingOperator(op1)
-	c.Assert(oc.OperatorCount(operator.OpRegion), Equals, uint64(1)) // 1:region 2:leader
-	c.Assert(oc.OperatorCount(operator.OpLeader), Equals, uint64(1))
-	op2 = newTestOperator(2, tc.GetRegion(2).GetRegionEpoch(), operator.OpRegion)
-	op2.SetPriorityLevel(core.HighPriority)
-	oc.AddWaitingOperator(op2)
-	c.Assert(oc.OperatorCount(operator.OpRegion), Equals, uint64(2)) // 1:region 2:region
-	c.Assert(oc.OperatorCount(operator.OpLeader), Equals, uint64(0))
+	{
+		op1 := newTestOperator(1, tc.GetRegion(1).GetRegionEpoch(), operator.OpRegion)
+		oc.AddWaitingOperator(op1)
+		c.Assert(oc.OperatorCount(operator.OpRegion), Equals, uint64(1)) // 1:region 2:leader
+		c.Assert(oc.OperatorCount(operator.OpLeader), Equals, uint64(1))
+		op2 := newTestOperator(2, tc.GetRegion(2).GetRegionEpoch(), operator.OpRegion)
+		op2.SetPriorityLevel(core.HighPriority)
+		oc.AddWaitingOperator(op2)
+		c.Assert(oc.OperatorCount(operator.OpRegion), Equals, uint64(2)) // 1:region 2:region
+		c.Assert(oc.OperatorCount(operator.OpLeader), Equals, uint64(0))
+	}
 }
 
 func (s *testOperatorControllerSuite) TestStoreOverloaded(c *C) {
@@ -940,10 +948,12 @@ func (s *testOperatorControllerSuite) TestStoreOverloaded(c *C) {
 	c.Assert(tc.addLeaderRegion(1, 2, 3, 4), IsNil)
 	region := tc.GetRegion(1).Clone(core.SetApproximateSize(60))
 	tc.putRegion(region)
-	op1 := lb.Schedule(tc)[0]
-	c.Assert(op1, NotNil)
-	c.Assert(oc.AddOperator(op1), IsTrue)
-	c.Assert(oc.RemoveOperator(op1), IsTrue)
+	{
+		op1 := lb.Schedule(tc)[0]
+		c.Assert(op1, NotNil)
+		c.Assert(oc.AddOperator(op1), IsTrue)
+		c.Assert(oc.RemoveOperator(op1), IsTrue)
+	}
 	for i := 0; i < 10; i++ {
 		c.Assert(lb.Schedule(tc), IsNil)
 	}
@@ -952,7 +962,7 @@ func (s *testOperatorControllerSuite) TestStoreOverloaded(c *C) {
 	// scheduling one time needs 1/10 seconds
 	oc.SetAllStoresLimit(10)
 	for i := 0; i < 10; i++ {
-		op1 = lb.Schedule(tc)[0]
+		op1 := lb.Schedule(tc)[0]
 		c.Assert(op1, NotNil)
 		c.Assert(oc.AddOperator(op1), IsTrue)
 		c.Assert(oc.RemoveOperator(op1), IsTrue)
@@ -1006,6 +1016,7 @@ type testScheduleControllerSuite struct {
 
 func (s *testScheduleControllerSuite) SetUpSuite(c *C) {
 	s.ctx, s.cancel = context.WithCancel(context.Background())
+	c.Assert(failpoint.Enable("github.com/pingcap/pd/server/schedule/unexpectedOperator", "return(true)"), IsNil)
 }
 
 func (s *testScheduleControllerSuite) TearDownSuite(c *C) {
@@ -1048,53 +1059,67 @@ func (s *testScheduleControllerSuite) TestController(c *C) {
 	// limit = 2
 	lb.limit = 2
 	// count = 0
-	c.Assert(sc.AllowSchedule(), IsTrue)
-	op1 := newTestOperator(1, tc.GetRegion(1).GetRegionEpoch(), operator.OpLeader)
-	c.Assert(oc.AddWaitingOperator(op1), IsTrue)
-	// count = 1
-	c.Assert(sc.AllowSchedule(), IsTrue)
-	op2 := newTestOperator(2, tc.GetRegion(2).GetRegionEpoch(), operator.OpLeader)
-	c.Assert(oc.AddWaitingOperator(op2), IsTrue)
-	// count = 2
-	c.Assert(sc.AllowSchedule(), IsFalse)
-	c.Assert(oc.RemoveOperator(op1), IsTrue)
-	// count = 1
-	c.Assert(sc.AllowSchedule(), IsTrue)
+	{
+		c.Assert(sc.AllowSchedule(), IsTrue)
+		op1 := newTestOperator(1, tc.GetRegion(1).GetRegionEpoch(), operator.OpLeader)
+		c.Assert(oc.AddWaitingOperator(op1), IsTrue)
+		// count = 1
+		c.Assert(sc.AllowSchedule(), IsTrue)
+		op2 := newTestOperator(2, tc.GetRegion(2).GetRegionEpoch(), operator.OpLeader)
+		c.Assert(oc.AddWaitingOperator(op2), IsTrue)
+		// count = 2
+		c.Assert(sc.AllowSchedule(), IsFalse)
+		c.Assert(oc.RemoveOperator(op1), IsTrue)
+		// count = 1
+		c.Assert(sc.AllowSchedule(), IsTrue)
+	}
 
+	op11 := newTestOperator(1, tc.GetRegion(1).GetRegionEpoch(), operator.OpLeader)
 	// add a PriorityKind operator will remove old operator
-	op3 := newTestOperator(2, tc.GetRegion(2).GetRegionEpoch(), operator.OpHotRegion)
-	op3.SetPriorityLevel(core.HighPriority)
-	c.Assert(oc.AddWaitingOperator(op1), IsTrue)
-	c.Assert(sc.AllowSchedule(), IsFalse)
-	c.Assert(oc.AddWaitingOperator(op3), IsTrue)
-	c.Assert(sc.AllowSchedule(), IsTrue)
-	c.Assert(oc.RemoveOperator(op3), IsTrue)
+	{
+		op3 := newTestOperator(2, tc.GetRegion(2).GetRegionEpoch(), operator.OpHotRegion)
+		op3.SetPriorityLevel(core.HighPriority)
+		c.Assert(oc.AddWaitingOperator(op11), IsTrue)
+		c.Assert(sc.AllowSchedule(), IsFalse)
+		c.Assert(oc.AddWaitingOperator(op3), IsTrue)
+		c.Assert(sc.AllowSchedule(), IsTrue)
+		c.Assert(oc.RemoveOperator(op3), IsTrue)
+	}
 
 	// add a admin operator will remove old operator
-	c.Assert(oc.AddWaitingOperator(op2), IsTrue)
-	c.Assert(sc.AllowSchedule(), IsFalse)
-	op4 := newTestOperator(2, tc.GetRegion(2).GetRegionEpoch(), operator.OpAdmin)
-	op4.SetPriorityLevel(core.HighPriority)
-	c.Assert(oc.AddWaitingOperator(op4), IsTrue)
-	c.Assert(sc.AllowSchedule(), IsTrue)
-	c.Assert(oc.RemoveOperator(op4), IsTrue)
+	{
+		op2 := newTestOperator(2, tc.GetRegion(2).GetRegionEpoch(), operator.OpLeader)
+		c.Assert(oc.AddWaitingOperator(op2), IsTrue)
+		c.Assert(sc.AllowSchedule(), IsFalse)
+		op4 := newTestOperator(2, tc.GetRegion(2).GetRegionEpoch(), operator.OpAdmin)
+		op4.SetPriorityLevel(core.HighPriority)
+		c.Assert(oc.AddWaitingOperator(op4), IsTrue)
+		c.Assert(sc.AllowSchedule(), IsTrue)
+		c.Assert(oc.RemoveOperator(op4), IsTrue)
+	}
 
 	// test wrong region id.
-	op5 := newTestOperator(3, &metapb.RegionEpoch{}, operator.OpHotRegion)
-	c.Assert(oc.AddWaitingOperator(op5), IsFalse)
+	{
+		op5 := newTestOperator(3, &metapb.RegionEpoch{}, operator.OpHotRegion)
+		c.Assert(oc.AddWaitingOperator(op5), IsFalse)
+	}
 
 	// test wrong region epoch.
-	c.Assert(oc.RemoveOperator(op1), IsTrue)
+	c.Assert(oc.RemoveOperator(op11), IsTrue)
 	epoch := &metapb.RegionEpoch{
 		Version: tc.GetRegion(1).GetRegionEpoch().GetVersion() + 1,
 		ConfVer: tc.GetRegion(1).GetRegionEpoch().GetConfVer(),
 	}
-	op6 := newTestOperator(1, epoch, operator.OpLeader)
-	c.Assert(oc.AddWaitingOperator(op6), IsFalse)
+	{
+		op6 := newTestOperator(1, epoch, operator.OpLeader)
+		c.Assert(oc.AddWaitingOperator(op6), IsFalse)
+	}
 	epoch.Version--
-	op6 = newTestOperator(1, epoch, operator.OpLeader)
-	c.Assert(oc.AddWaitingOperator(op6), IsTrue)
-	c.Assert(oc.RemoveOperator(op6), IsTrue)
+	{
+		op6 := newTestOperator(1, epoch, operator.OpLeader)
+		c.Assert(oc.AddWaitingOperator(op6), IsTrue)
+		c.Assert(oc.RemoveOperator(op6), IsTrue)
+	}
 }
 
 func (s *testScheduleControllerSuite) TestInterval(c *C) {
diff --git a/server/schedule/checker/merge_checker_test.go b/server/schedule/checker/merge_checker_test.go
index 77f54ea09..8229d0f31 100644
--- a/server/schedule/checker/merge_checker_test.go
+++ b/server/schedule/checker/merge_checker_test.go
@@ -145,13 +145,6 @@ func (s *testMergeCheckerSuite) TestBasic(c *C) {
 	c.Assert(ops, IsNil)
 	ops = s.mc.Check(s.regions[2])
 	c.Assert(ops, NotNil)
-	for _, op := range ops {
-		op.SetStartTime(time.Now())
-		op.SetStartTime(op.GetStartTime().Add(-operator.LeaderOperatorWaitTime - time.Second))
-		c.Assert(op.IsTimeout(), IsFalse)
-		op.SetStartTime(op.GetStartTime().Add(-operator.RegionOperatorWaitTime - time.Second))
-		c.Assert(op.IsTimeout(), IsTrue)
-	}
 	// Check merge with previous region.
 	c.Assert(ops[0].RegionID(), Equals, s.regions[2].GetID())
 	c.Assert(ops[1].RegionID(), Equals, s.regions[1].GetID())
diff --git a/server/schedule/operator/operator_test.go b/server/schedule/operator/operator_test.go
index 72d4ecefa..93dcb3621 100644
--- a/server/schedule/operator/operator_test.go
+++ b/server/schedule/operator/operator_test.go
@@ -140,11 +140,11 @@ func (s *testOperatorSuite) TestOperator(c *C) {
 	op := s.newTestOperator(1, OpLeader|OpRegion, steps...)
 	c.Assert(op.GetPriorityLevel(), Equals, core.HighPriority)
 	s.checkSteps(c, op, steps)
+	op.Start()
 	c.Assert(op.Check(region), IsNil)
-	c.Assert(op.IsFinish(), IsTrue)
-	op.startTime = time.Now()
-	op.startTime = op.startTime.Add(-RegionOperatorWaitTime - time.Second)
-	c.Assert(op.IsTimeout(), IsFalse)
+	c.Assert(op.Status(), Equals, SUCCESS)
+	SetOperatorStatusReachTime(op, STARTED, time.Now().Add(-RegionOperatorWaitTime-time.Second))
+	c.Assert(op.CheckTimeout(), IsFalse)
 
 	// addPeer1, transferLeader1, removePeer2
 	steps = []OpStep{
@@ -154,14 +154,14 @@ func (s *testOperatorSuite) TestOperator(c *C) {
 	}
 	op = s.newTestOperator(1, OpLeader|OpRegion, steps...)
 	s.checkSteps(c, op, steps)
+	op.Start()
 	c.Assert(op.Check(region), Equals, RemovePeer{FromStore: 2})
 	c.Assert(atomic.LoadInt32(&op.currentStep), Equals, int32(2))
-	op.startTime = time.Now()
-	c.Assert(op.IsTimeout(), IsFalse)
-	op.startTime = op.startTime.Add(-LeaderOperatorWaitTime - time.Second)
-	c.Assert(op.IsTimeout(), IsFalse)
-	op.startTime = op.startTime.Add(-RegionOperatorWaitTime - time.Second)
-	c.Assert(op.IsTimeout(), IsTrue)
+	c.Assert(op.CheckTimeout(), IsFalse)
+	SetOperatorStatusReachTime(op, STARTED, op.GetStartTime().Add(-LeaderOperatorWaitTime-time.Second))
+	c.Assert(op.CheckTimeout(), IsFalse)
+	SetOperatorStatusReachTime(op, STARTED, op.GetStartTime().Add(-RegionOperatorWaitTime-time.Second))
+	c.Assert(op.CheckTimeout(), IsTrue)
 	res, err := json.Marshal(op)
 	c.Assert(err, IsNil)
 	c.Assert(len(res), Equals, len(op.String())+2)
@@ -169,10 +169,10 @@ func (s *testOperatorSuite) TestOperator(c *C) {
 	// check short timeout for transfer leader only operators.
 	steps = []OpStep{TransferLeader{FromStore: 2, ToStore: 1}}
 	op = s.newTestOperator(1, OpLeader, steps...)
-	op.startTime = time.Now()
-	c.Assert(op.IsTimeout(), IsFalse)
-	op.startTime = op.startTime.Add(-LeaderOperatorWaitTime - time.Second)
-	c.Assert(op.IsTimeout(), IsTrue)
+	op.Start()
+	c.Assert(op.CheckTimeout(), IsFalse)
+	SetOperatorStatusReachTime(op, STARTED, op.GetStartTime().Add(-LeaderOperatorWaitTime-time.Second))
+	c.Assert(op.CheckTimeout(), IsTrue)
 }
 
 func (s *testOperatorSuite) TestInfluence(c *C) {
@@ -267,3 +267,142 @@ func (s *testOperatorSuite) TestOperatorKind(c *C) {
 	_, err = ParseOperatorKind("foobar")
 	c.Assert(err, NotNil)
 }
+
+func (s *testOperatorSuite) TestCheckSuccess(c *C) {
+	{
+		steps := []OpStep{
+			AddPeer{ToStore: 1, PeerID: 1},
+			TransferLeader{FromStore: 2, ToStore: 1},
+			RemovePeer{FromStore: 2},
+		}
+		op := s.newTestOperator(1, OpLeader|OpRegion, steps...)
+		c.Assert(op.Status(), Equals, CREATED)
+		c.Assert(op.CheckSuccess(), IsFalse)
+		c.Assert(op.Start(), IsTrue)
+		c.Assert(op.CheckSuccess(), IsFalse)
+		op.currentStep = int32(len(op.steps))
+		c.Assert(op.CheckSuccess(), IsTrue)
+		c.Assert(op.CheckSuccess(), IsTrue)
+	}
+	{
+		steps := []OpStep{
+			AddPeer{ToStore: 1, PeerID: 1},
+			TransferLeader{FromStore: 2, ToStore: 1},
+			RemovePeer{FromStore: 2},
+		}
+		op := s.newTestOperator(1, OpLeader|OpRegion, steps...)
+		op.currentStep = int32(len(op.steps))
+		c.Assert(op.Status(), Equals, CREATED)
+		c.Assert(op.CheckSuccess(), IsFalse)
+		c.Assert(op.Start(), IsTrue)
+		c.Assert(op.CheckSuccess(), IsTrue)
+		c.Assert(op.CheckSuccess(), IsTrue)
+	}
+}
+
+func (s *testOperatorSuite) TestCheckTimeout(c *C) {
+	{
+		steps := []OpStep{
+			AddPeer{ToStore: 1, PeerID: 1},
+			TransferLeader{FromStore: 2, ToStore: 1},
+			RemovePeer{FromStore: 2},
+		}
+		op := s.newTestOperator(1, OpLeader|OpRegion, steps...)
+		c.Assert(op.Status(), Equals, CREATED)
+		c.Assert(op.Start(), IsTrue)
+		op.currentStep = int32(len(op.steps))
+		c.Assert(op.CheckTimeout(), IsFalse)
+		c.Assert(op.Status(), Equals, SUCCESS)
+	}
+	{
+		steps := []OpStep{
+			AddPeer{ToStore: 1, PeerID: 1},
+			TransferLeader{FromStore: 2, ToStore: 1},
+			RemovePeer{FromStore: 2},
+		}
+		op := s.newTestOperator(1, OpLeader|OpRegion, steps...)
+		c.Assert(op.Status(), Equals, CREATED)
+		c.Assert(op.Start(), IsTrue)
+		op.currentStep = int32(len(op.steps))
+		SetOperatorStatusReachTime(op, STARTED, time.Now().Add(-RegionOperatorWaitTime))
+		c.Assert(op.CheckTimeout(), IsFalse)
+		c.Assert(op.Status(), Equals, SUCCESS)
+	}
+}
+
+func (s *testOperatorSuite) TestStart(c *C) {
+	steps := []OpStep{
+		AddPeer{ToStore: 1, PeerID: 1},
+		TransferLeader{FromStore: 2, ToStore: 1},
+		RemovePeer{FromStore: 2},
+	}
+	op := s.newTestOperator(1, OpLeader|OpRegion, steps...)
+	c.Assert(op.stepTime, Equals, int64(0))
+	c.Assert(op.Status(), Equals, CREATED)
+	c.Assert(op.Start(), IsTrue)
+	c.Assert(op.stepTime, Not(Equals), 0)
+	c.Assert(op.Status(), Equals, STARTED)
+}
+
+func (s *testOperatorSuite) TestCheckExpired(c *C) {
+	steps := []OpStep{
+		AddPeer{ToStore: 1, PeerID: 1},
+		TransferLeader{FromStore: 2, ToStore: 1},
+		RemovePeer{FromStore: 2},
+	}
+	op := s.newTestOperator(1, OpLeader|OpRegion, steps...)
+	c.Assert(op.CheckExpired(), IsFalse)
+	c.Assert(op.Status(), Equals, CREATED)
+	SetOperatorStatusReachTime(op, CREATED, time.Now().Add(-OperatorExpireTime))
+	c.Assert(op.CheckExpired(), IsTrue)
+	c.Assert(op.Status(), Equals, EXPIRED)
+}
+
+func (s *testOperatorSuite) TestCheck(c *C) {
+	{
+		region := s.newTestRegion(1, 1, [2]uint64{1, 1}, [2]uint64{2, 2})
+		steps := []OpStep{
+			AddPeer{ToStore: 1, PeerID: 1},
+			TransferLeader{FromStore: 2, ToStore: 1},
+			RemovePeer{FromStore: 2},
+		}
+		op := s.newTestOperator(1, OpLeader|OpRegion, steps...)
+		c.Assert(op.Start(), IsTrue)
+		c.Assert(op.Check(region), NotNil)
+		c.Assert(op.Status(), Equals, STARTED)
+		region = s.newTestRegion(1, 1, [2]uint64{1, 1})
+		c.Assert(op.Check(region), IsNil)
+		c.Assert(op.Status(), Equals, SUCCESS)
+	}
+	{
+		region := s.newTestRegion(1, 1, [2]uint64{1, 1}, [2]uint64{2, 2})
+		steps := []OpStep{
+			AddPeer{ToStore: 1, PeerID: 1},
+			TransferLeader{FromStore: 2, ToStore: 1},
+			RemovePeer{FromStore: 2},
+		}
+		op := s.newTestOperator(1, OpLeader|OpRegion, steps...)
+		c.Assert(op.Start(), IsTrue)
+		c.Assert(op.Check(region), NotNil)
+		c.Assert(op.Status(), Equals, STARTED)
+		op.status.setTime(STARTED, time.Now().Add(-RegionOperatorWaitTime))
+		c.Assert(op.Check(region), NotNil)
+		c.Assert(op.Status(), Equals, TIMEOUT)
+	}
+	{
+		region := s.newTestRegion(1, 1, [2]uint64{1, 1}, [2]uint64{2, 2})
+		steps := []OpStep{
+			AddPeer{ToStore: 1, PeerID: 1},
+			TransferLeader{FromStore: 2, ToStore: 1},
+			RemovePeer{FromStore: 2},
+		}
+		op := s.newTestOperator(1, OpLeader|OpRegion, steps...)
+		c.Assert(op.Start(), IsTrue)
+		c.Assert(op.Check(region), NotNil)
+		c.Assert(op.Status(), Equals, STARTED)
+		op.status.setTime(STARTED, time.Now().Add(-RegionOperatorWaitTime))
+		region = s.newTestRegion(1, 1, [2]uint64{1, 1})
+		c.Assert(op.Check(region), IsNil)
+		c.Assert(op.Status(), Equals, SUCCESS)
+	}
+}
diff --git a/server/schedule/operator/status_test.go b/server/schedule/operator/status_test.go
new file mode 100644
index 000000000..92b508541
--- /dev/null
+++ b/server/schedule/operator/status_test.go
@@ -0,0 +1,34 @@
+// Copyright 2019 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package operator
+
+import (
+	. "github.com/pingcap/check"
+)
+
+var _ = Suite(&testOpStatusSuite{})
+
+type testOpStatusSuite struct{}
+
+func (s *testOpStatusSuite) TestIsEndStatus(c *C) {
+	for st := OpStatus(0); st < firstEndStatus; st++ {
+		c.Assert(IsEndStatus(st), IsFalse)
+	}
+	for st := firstEndStatus; st < statusCount; st++ {
+		c.Assert(IsEndStatus(st), IsTrue)
+	}
+	for st := statusCount; st < statusCount+100; st++ {
+		c.Assert(IsEndStatus(st), IsFalse)
+	}
+}
diff --git a/server/schedule/operator/status_tracker_test.go b/server/schedule/operator/status_tracker_test.go
new file mode 100644
index 000000000..3c65034d0
--- /dev/null
+++ b/server/schedule/operator/status_tracker_test.go
@@ -0,0 +1,184 @@
+// Copyright 2019 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package operator
+
+import (
+	"time"
+
+	. "github.com/pingcap/check"
+)
+
+var _ = Suite(&testOpStatusTrackerSuite{})
+
+type testOpStatusTrackerSuite struct{}
+
+func (s *testOpStatusTrackerSuite) TestCreate(c *C) {
+	before := time.Now()
+	trk := NewOpStatusTracker()
+	c.Assert(trk.Status(), Equals, CREATED)
+	c.Assert(trk.ReachTime(), DeepEquals, trk.ReachTimeOf(CREATED))
+	checkTimeOrder(c, before, trk.ReachTime(), time.Now())
+	checkReachTime(c, &trk, CREATED)
+}
+
+func (s *testOpStatusTrackerSuite) TestNonEndTrans(c *C) {
+	{
+		trk := NewOpStatusTracker()
+		checkInvalidTrans(c, &trk, SUCCESS, REPLACED, TIMEOUT)
+		checkValidTrans(c, &trk, STARTED)
+		checkInvalidTrans(c, &trk, EXPIRED)
+		checkValidTrans(c, &trk, SUCCESS)
+		checkReachTime(c, &trk, CREATED, STARTED, SUCCESS)
+	}
+	{
+		trk := NewOpStatusTracker()
+		checkValidTrans(c, &trk, CANCELED)
+		checkReachTime(c, &trk, CREATED, CANCELED)
+	}
+	{
+		trk := NewOpStatusTracker()
+		checkValidTrans(c, &trk, STARTED)
+		checkValidTrans(c, &trk, CANCELED)
+		checkReachTime(c, &trk, CREATED, STARTED, CANCELED)
+	}
+	{
+		trk := NewOpStatusTracker()
+		checkValidTrans(c, &trk, STARTED)
+		checkValidTrans(c, &trk, REPLACED)
+		checkReachTime(c, &trk, CREATED, STARTED, REPLACED)
+	}
+	{
+		trk := NewOpStatusTracker()
+		checkValidTrans(c, &trk, EXPIRED)
+		checkReachTime(c, &trk, CREATED, EXPIRED)
+	}
+	{
+		trk := NewOpStatusTracker()
+		checkValidTrans(c, &trk, STARTED)
+		checkValidTrans(c, &trk, TIMEOUT)
+		checkReachTime(c, &trk, CREATED, STARTED, TIMEOUT)
+	}
+}
+
+func (s *testOpStatusTrackerSuite) TestEndStatusTrans(c *C) {
+	allStatus := make([]OpStatus, 0, statusCount)
+	for st := OpStatus(0); st < statusCount; st++ {
+		allStatus = append(allStatus, st)
+	}
+	for from := firstEndStatus; from < statusCount; from++ {
+		trk := NewOpStatusTracker()
+		trk.current = from
+		c.Assert(trk.IsEnd(), IsTrue)
+		checkInvalidTrans(c, &trk, allStatus...)
+	}
+}
+
+func (s *testOpStatusTrackerSuite) TestCheckExpired(c *C) {
+	{
+		// Not expired
+		before := time.Now()
+		trk := NewOpStatusTracker()
+		after := time.Now()
+		c.Assert(trk.CheckExpired(10*time.Second), IsFalse)
+		c.Assert(trk.Status(), Equals, CREATED)
+		checkTimeOrder(c, before, trk.ReachTime(), after)
+	}
+	{
+		// Expired but status not changed
+		trk := NewOpStatusTracker()
+		trk.setTime(CREATED, time.Now().Add(-10*time.Second))
+		c.Assert(trk.CheckExpired(5*time.Second), IsTrue)
+		c.Assert(trk.Status(), Equals, EXPIRED)
+	}
+	{
+		// Expired and status changed
+		trk := NewOpStatusTracker()
+		before := time.Now()
+		c.Assert(trk.To(EXPIRED), IsTrue)
+		after := time.Now()
+		c.Assert(trk.CheckExpired(0), IsTrue)
+		c.Assert(trk.Status(), Equals, EXPIRED)
+		checkTimeOrder(c, before, trk.ReachTime(), after)
+	}
+}
+
+func (s *testOpStatusTrackerSuite) TestCheckTimeout(c *C) {
+	{
+		// Not timeout
+		trk := NewOpStatusTracker()
+		before := time.Now()
+		c.Assert(trk.To(STARTED), IsTrue)
+		after := time.Now()
+		c.Assert(trk.CheckTimeout(10*time.Second), IsFalse)
+		c.Assert(trk.Status(), Equals, STARTED)
+		checkTimeOrder(c, before, trk.ReachTime(), after)
+	}
+	{
+		// Timeout but status not changed
+		trk := NewOpStatusTracker()
+		c.Assert(trk.To(STARTED), IsTrue)
+		trk.setTime(STARTED, time.Now().Add(-10*time.Second))
+		c.Assert(trk.CheckTimeout(5*time.Second), IsTrue)
+		c.Assert(trk.Status(), Equals, TIMEOUT)
+	}
+	{
+		// Timeout and status changed
+		trk := NewOpStatusTracker()
+		c.Assert(trk.To(STARTED), IsTrue)
+		before := time.Now()
+		c.Assert(trk.To(TIMEOUT), IsTrue)
+		after := time.Now()
+		c.Assert(trk.CheckTimeout(0), IsTrue)
+		c.Assert(trk.Status(), Equals, TIMEOUT)
+		checkTimeOrder(c, before, trk.ReachTime(), after)
+	}
+}
+
+func checkTimeOrder(c *C, t1, t2, t3 time.Time) {
+	c.Assert(t1.Before(t2), IsTrue)
+	c.Assert(t3.After(t2), IsTrue)
+}
+
+func checkValidTrans(c *C, trk *OpStatusTracker, st OpStatus) {
+	before := time.Now()
+	c.Assert(trk.To(st), IsTrue)
+	c.Assert(trk.Status(), Equals, st)
+	c.Assert(trk.ReachTime(), DeepEquals, trk.ReachTimeOf(st))
+	checkTimeOrder(c, before, trk.ReachTime(), time.Now())
+}
+
+func checkInvalidTrans(c *C, trk *OpStatusTracker, sts ...OpStatus) {
+	origin := trk.Status()
+	originTime := trk.ReachTime()
+	sts = append(sts, statusCount, statusCount+1, statusCount+10)
+	for _, st := range sts {
+		c.Assert(trk.To(st), IsFalse)
+		c.Assert(trk.Status(), Equals, origin)
+		c.Assert(trk.ReachTime(), DeepEquals, originTime)
+	}
+}
+
+func checkReachTime(c *C, trk *OpStatusTracker, reached ...OpStatus) {
+	reachedMap := make(map[OpStatus]struct{}, len(reached))
+	for _, st := range reached {
+		c.Assert(trk.ReachTimeOf(st).IsZero(), IsFalse)
+		reachedMap[st] = struct{}{}
+	}
+	for st := OpStatus(0); st <= statusCount+10; st++ {
+		if _, ok := reachedMap[st]; ok {
+			continue
+		}
+		c.Assert(trk.ReachTimeOf(st).IsZero(), IsTrue)
+	}
+}
diff --git a/server/schedule/operator/test_util.go b/server/schedule/operator/test_util.go
new file mode 100644
index 000000000..e6d6c1c60
--- /dev/null
+++ b/server/schedule/operator/test_util.go
@@ -0,0 +1,24 @@
+// Copyright 2019 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package operator
+
+import (
+	"time"
+)
+
+// SetOperatorStatusReachTime sets the reach time of the operator.
+// NOTE: Should only use in test.
+func SetOperatorStatusReachTime(op *Operator, st OpStatus, t time.Time) {
+	op.status.setTime(st, t)
+}
diff --git a/server/schedule/operator_controller_test.go b/server/schedule/operator_controller_test.go
index 85377f542..5747c4b60 100644
--- a/server/schedule/operator_controller_test.go
+++ b/server/schedule/operator_controller_test.go
@@ -45,6 +45,7 @@ type testOperatorControllerSuite struct {
 
 func (t *testOperatorControllerSuite) SetUpSuite(c *C) {
 	t.ctx, t.cancel = context.WithCancel(context.Background())
+	c.Assert(failpoint.Enable("github.com/pingcap/pd/server/schedule/unexpectedOperator", "return(true)"), IsNil)
 }
 
 func (t *testOperatorControllerSuite) TearDownSuite(c *C) {
@@ -64,10 +65,12 @@ func (t *testOperatorControllerSuite) TestGetOpInfluence(c *C) {
 	}
 	op1 := operator.NewOperator("test", "test", 1, &metapb.RegionEpoch{}, operator.OpRegion, steps...)
 	op2 := operator.NewOperator("test", "test", 2, &metapb.RegionEpoch{}, operator.OpRegion, steps...)
+	c.Assert(op1.Start(), IsTrue)
 	oc.SetOperator(op1)
+	c.Assert(op2.Start(), IsTrue)
 	oc.SetOperator(op2)
 	go func(ctx context.Context) {
-		c.Assert(oc.RemoveOperator(op1), IsTrue)
+		checkRemoveOperatorSuccess(c, oc, op1)
 		for {
 			select {
 			case <-ctx.Done():
@@ -107,13 +110,13 @@ func (t *testOperatorControllerSuite) TestOperatorStatus(c *C) {
 	op2 := operator.NewOperator("test", "test", 2, &metapb.RegionEpoch{}, operator.OpRegion, steps...)
 	region1 := tc.GetRegion(1)
 	region2 := tc.GetRegion(2)
-	op1.SetStartTime(time.Now())
+	c.Assert(op1.Start(), IsTrue)
 	oc.SetOperator(op1)
-	op2.SetStartTime(time.Now())
+	c.Assert(op2.Start(), IsTrue)
 	oc.SetOperator(op2)
 	c.Assert(oc.GetOperatorStatus(1).Status, Equals, pdpb.OperatorStatus_RUNNING)
 	c.Assert(oc.GetOperatorStatus(2).Status, Equals, pdpb.OperatorStatus_RUNNING)
-	op1.SetStartTime(time.Now().Add(-10 * time.Minute))
+	operator.SetOperatorStatusReachTime(op1, operator.STARTED, time.Now().Add(-10*time.Minute))
 	region2 = ApplyOperatorStep(region2, op2)
 	tc.PutRegion(region2)
 	oc.Dispatch(region1, "test")
@@ -125,6 +128,69 @@ func (t *testOperatorControllerSuite) TestOperatorStatus(c *C) {
 	c.Assert(oc.GetOperatorStatus(2).Status, Equals, pdpb.OperatorStatus_SUCCESS)
 }
 
+func (t *testOperatorControllerSuite) TestCheckAddUnexpectedStatus(c *C) {
+	c.Assert(failpoint.Disable("github.com/pingcap/pd/server/schedule/unexpectedOperator"), IsNil)
+	opt := mockoption.NewScheduleOptions()
+	tc := mockcluster.NewCluster(opt)
+	oc := NewOperatorController(t.ctx, tc, mockhbstream.NewHeartbeatStream())
+	tc.AddLeaderStore(1, 0)
+	tc.AddLeaderStore(2, 1)
+	tc.AddLeaderRegion(1, 2, 1)
+	tc.AddLeaderRegion(2, 2, 1)
+	region1 := tc.GetRegion(1)
+	steps := []operator.OpStep{
+		operator.RemovePeer{FromStore: 1},
+		operator.AddPeer{ToStore: 1, PeerID: 4},
+	}
+	{
+		// finished op
+		op := operator.NewOperator("test", "test", 1, &metapb.RegionEpoch{}, operator.OpRegion, operator.TransferLeader{ToStore: 2})
+		c.Assert(oc.checkAddOperator(op), IsTrue)
+		op.Start()
+		c.Assert(oc.checkAddOperator(op), IsFalse) // started
+		c.Assert(op.Check(region1), IsNil)
+		c.Assert(op.Status(), Equals, operator.SUCCESS)
+		c.Assert(oc.checkAddOperator(op), IsFalse) // success
+	}
+	{
+		// finished op canceled
+		op := operator.NewOperator("test", "test", 1, &metapb.RegionEpoch{}, operator.OpRegion, operator.TransferLeader{ToStore: 2})
+		c.Assert(oc.checkAddOperator(op), IsTrue)
+		c.Assert(op.Cancel(), IsTrue)
+		c.Assert(oc.checkAddOperator(op), IsFalse)
+	}
+	{
+		// finished op replaced
+		op := operator.NewOperator("test", "test", 1, &metapb.RegionEpoch{}, operator.OpRegion, operator.TransferLeader{ToStore: 2})
+		c.Assert(oc.checkAddOperator(op), IsTrue)
+		c.Assert(op.Start(), IsTrue)
+		c.Assert(op.Replace(), IsTrue)
+		c.Assert(oc.checkAddOperator(op), IsFalse)
+	}
+	{
+		// finished op expired
+		op1 := operator.NewOperator("test", "test", 1, &metapb.RegionEpoch{}, operator.OpRegion, operator.TransferLeader{ToStore: 2})
+		op2 := operator.NewOperator("test", "test", 2, &metapb.RegionEpoch{}, operator.OpRegion, operator.TransferLeader{ToStore: 1})
+		c.Assert(oc.checkAddOperator(op1, op2), IsTrue)
+		operator.SetOperatorStatusReachTime(op1, operator.CREATED, time.Now().Add(-operator.OperatorExpireTime))
+		operator.SetOperatorStatusReachTime(op2, operator.CREATED, time.Now().Add(-operator.OperatorExpireTime))
+		c.Assert(oc.checkAddOperator(op1, op2), IsFalse)
+		c.Assert(op1.Status(), Equals, operator.EXPIRED)
+		c.Assert(op2.Status(), Equals, operator.EXPIRED)
+	}
+	// finished op never timeout
+
+	{
+		// unfinished op timeout
+		op := operator.NewOperator("test", "test", 1, &metapb.RegionEpoch{}, operator.OpRegion, steps...)
+		c.Assert(oc.checkAddOperator(op), IsTrue)
+		op.Start()
+		operator.SetOperatorStatusReachTime(op, operator.STARTED, time.Now().Add(-operator.RegionOperatorWaitTime))
+		c.Assert(op.CheckTimeout(), IsTrue)
+		c.Assert(oc.checkAddOperator(op), IsFalse)
+	}
+}
+
 // issue #1716
 func (t *testOperatorControllerSuite) TestConcurrentRemoveOperator(c *C) {
 	opt := mockoption.NewScheduleOptions()
@@ -143,6 +209,7 @@ func (t *testOperatorControllerSuite) TestConcurrentRemoveOperator(c *C) {
 	// unfinished op with high priority
 	op2 := operator.NewOperator("test", "test", 1, &metapb.RegionEpoch{}, operator.OpRegion|operator.OpAdmin, steps...)
 
+	c.Assert(op1.Start(), IsTrue)
 	oc.SetOperator(op1)
 
 	c.Assert(failpoint.Enable("github.com/pingcap/pd/server/schedule/concurrentRemoveOperator", "return(true)"), IsNil)
@@ -187,9 +254,13 @@ func (t *testOperatorControllerSuite) TestPollDispatchRegion(c *C) {
 	region4 := tc.GetRegion(4)
 	// Adds operator and pushes to the notifier queue.
 	{
+		c.Assert(op1.Start(), IsTrue)
 		oc.SetOperator(op1)
+		c.Assert(op3.Start(), IsTrue)
 		oc.SetOperator(op3)
+		c.Assert(op4.Start(), IsTrue)
 		oc.SetOperator(op4)
+		c.Assert(op2.Start(), IsTrue)
 		oc.SetOperator(op2)
 		heap.Push(&oc.opNotifierQueue, &operatorWithTime{op: op1, time: time.Now().Add(100 * time.Millisecond)})
 		heap.Push(&oc.opNotifierQueue, &operatorWithTime{op: op3, time: time.Now().Add(300 * time.Millisecond)})
@@ -246,7 +317,7 @@ func (t *testOperatorControllerSuite) TestStoreLimit(c *C) {
 	for i := uint64(1); i <= 5; i++ {
 		op := operator.NewOperator("test", "test", 1, &metapb.RegionEpoch{}, operator.OpRegion, operator.AddPeer{ToStore: 2, PeerID: i})
 		c.Assert(oc.AddOperator(op), IsTrue)
-		c.Assert(oc.RemoveOperator(op), IsTrue)
+		checkRemoveOperatorSuccess(c, oc, op)
 	}
 	op := operator.NewOperator("test", "test", 1, &metapb.RegionEpoch{}, operator.OpRegion, operator.AddPeer{ToStore: 2, PeerID: 1})
 	c.Assert(oc.AddOperator(op), IsFalse)
@@ -256,13 +327,13 @@ func (t *testOperatorControllerSuite) TestStoreLimit(c *C) {
 	for i := uint64(1); i <= 10; i++ {
 		op = operator.NewOperator("test", "test", i, &metapb.RegionEpoch{}, operator.OpRegion, operator.AddPeer{ToStore: 2, PeerID: i})
 		c.Assert(oc.AddOperator(op), IsTrue)
-		c.Assert(oc.RemoveOperator(op), IsTrue)
+		checkRemoveOperatorSuccess(c, oc, op)
 	}
 	oc.SetAllStoresLimit(1)
 	for i := uint64(1); i <= 5; i++ {
 		op = operator.NewOperator("test", "test", i, &metapb.RegionEpoch{}, operator.OpRegion, operator.AddPeer{ToStore: 2, PeerID: i})
 		c.Assert(oc.AddOperator(op), IsTrue)
-		c.Assert(oc.RemoveOperator(op), IsTrue)
+		checkRemoveOperatorSuccess(c, oc, op)
 	}
 	op = operator.NewOperator("test", "test", 1, &metapb.RegionEpoch{}, operator.OpRegion, operator.AddPeer{ToStore: 2, PeerID: 1})
 	c.Assert(oc.AddOperator(op), IsFalse)
@@ -464,10 +535,10 @@ func (t *testOperatorControllerSuite) TestStoreLimitWithMerge(c *C) {
 		core.WithLeader(&metapb.Peer{Id: 109, StoreId: 2}),
 	)
 	tc.PutRegion(regions[2])
-	ops := mc.Check(regions[2])
-	c.Assert(ops, NotNil)
 	// The size of Region is less or equal than 1MB.
 	for i := 0; i < 50; i++ {
+		ops := mc.Check(regions[2])
+		c.Assert(ops, NotNil)
 		c.Assert(oc.AddOperator(ops...), IsTrue)
 		for _, op := range ops {
 			oc.RemoveOperator(op)
@@ -478,16 +549,20 @@ func (t *testOperatorControllerSuite) TestStoreLimitWithMerge(c *C) {
 		core.SetApproximateKeys(2),
 	)
 	tc.PutRegion(regions[2])
-	ops = mc.Check(regions[2])
-	c.Assert(ops, NotNil)
 	// The size of Region is more than 1MB but no more than 20MB.
 	for i := 0; i < 5; i++ {
+		ops := mc.Check(regions[2])
+		c.Assert(ops, NotNil)
 		c.Assert(oc.AddOperator(ops...), IsTrue)
 		for _, op := range ops {
 			oc.RemoveOperator(op)
 		}
 	}
-	c.Assert(oc.AddOperator(ops...), IsFalse)
+	{
+		ops := mc.Check(regions[2])
+		c.Assert(ops, NotNil)
+		c.Assert(oc.AddOperator(ops...), IsFalse)
+	}
 }
 
 func newRegionInfo(id uint64, startKey, endKey string, size, keys int64, leader []uint64, peers ...[]uint64) *core.RegionInfo {
@@ -507,3 +582,9 @@ func newRegionInfo(id uint64, startKey, endKey string, size, keys int64, leader
 		core.SetApproximateKeys(keys),
 	)
 }
+
+func checkRemoveOperatorSuccess(c *C, oc *OperatorController, op *operator.Operator) {
+	c.Assert(oc.RemoveOperator(op), IsTrue)
+	c.Assert(op.IsEnd(), IsTrue)
+	c.Assert(oc.GetOperatorStatus(op.RegionID()).Op, DeepEquals, op)
+}
diff --git a/server/schedule/test_util.go b/server/schedule/test_util.go
index c0dc4f3fc..a4ffcab34 100644
--- a/server/schedule/test_util.go
+++ b/server/schedule/test_util.go
@@ -22,6 +22,7 @@ import (
 
 // ApplyOperatorStep applies operator step. Only for test purpose.
 func ApplyOperatorStep(region *core.RegionInfo, op *operator.Operator) *core.RegionInfo {
+	_ = op.Start()
 	if step := op.Check(region); step != nil {
 		switch s := step.(type) {
 		case operator.TransferLeader:
@@ -92,7 +93,7 @@ func ApplyOperatorStep(region *core.RegionInfo, op *operator.Operator) *core.Reg
 func ApplyOperator(mc *mockcluster.Cluster, op *operator.Operator) {
 	origin := mc.GetRegion(op.RegionID())
 	region := origin
-	for !op.IsFinish() {
+	for !op.IsEnd() {
 		region = ApplyOperatorStep(region, op)
 	}
 	mc.PutRegion(region)