address comments

service/history/historyEngine.go

@@ -79,6 +79,8 @@ type (
 var _ Engine = (*historyEngineImpl)(nil)
 
 var (
+	// ErrTaskRetry is the error indicating that the timer / transfer task should be retried.
+	ErrTaskRetry = errors.New("passive task should retry due to condition in mutable state is not met")
 	// ErrDuplicate is exported temporarily for integration test
 	ErrDuplicate = errors.New("Duplicate task, completing it")
 	// ErrConflict is exported temporarily for integration test

service/history/historyEngineInterfaces.go

@@ -45,10 +45,6 @@ type (
 		timestamp         time.Time
 	}
 
-	// error which will be thrown if the timer / transfer task should be
-	// retries due to various of reasons
-	taskRetryError struct{}
-
 	// Engine represents an interface for managing workflow execution history.
 	Engine interface {
 		common.Daemon
@@ -140,12 +136,3 @@ type (
 		UnwatchHistoryEvent(identifier *workflowIdentifier, subscriberID string) error
 	}
 )
-
-// newTaskRetryError create a error which indicate the task should be retry
-func newTaskRetryError() *taskRetryError {
-	return &taskRetryError{}
-}
-
-func (e *taskRetryError) Error() string {
-	return "passive task should retry due to condition in mutable state is not met."
-}

service/history/timerQueueAckMgr.go

@@ -170,7 +170,6 @@ func (t *timerQueueAckMgrImpl) readTimerTasks() ([]*persistence.TimerTaskInfo, *
 		t.isReadFinished = true
 	}
 
-	// fillin the retry task
 	filteredTasks := []*persistence.TimerTaskInfo{}
 
 	// since we have already checked that the clusterName is a valid key of clusterReadLevel

service/history/timerQueueProcessorBase.go

@@ -179,7 +179,7 @@ func (t *timerQueueProcessorBase) processTaskWorker(workerWG *sync.WaitGroup, no
 
 				err := t.timerProcessor.process(task)
 				if err != nil {
-					if _, ok := err.(*taskRetryError); ok {
+					if err == ErrTaskRetry {
 						<-notificationChan
 					} else {
 						// We will retry until we don't find the timer task any more.

service/history/timerQueueStandbyProcessor.go

@@ -197,7 +197,7 @@ func (t *timerQueueStandbyProcessorImpl) processExpiredUserTimer(timerTask *pers
 				//
 				// we do not need to notity new timer to base, since if there is no new event being replicated
 				// checking again if the timer can be completed is meaningless
-				return newTaskRetryError()
+				return ErrTaskRetry
 			}
 			// since the user timer are already sorted, so if there is one timer which will not expired
 			// all user timer after this timer will not expired
@@ -237,7 +237,7 @@ func (t *timerQueueStandbyProcessorImpl) processActivityTimeout(timerTask *persi
 				//
 				// we do not need to notity new timer to base, since if there is no new event being replicated
 				// checking again if the timer can be completed is meaningless
-				return newTaskRetryError()
+				return ErrTaskRetry
 			}
 			// since the activity timer are already sorted, so if there is one timer which will not expired
 			// all activity timer after this timer will not expired
@@ -262,7 +262,7 @@ func (t *timerQueueStandbyProcessorImpl) processDecisionTimeout(timerTask *persi
 			//
 			// we do not need to notity new timer to base, since if there is no new event being replicated
 			// checking again if the timer can be completed is meaningless
-			return newTaskRetryError()
+			return ErrTaskRetry
 		}
 		return nil
 	})
@@ -276,7 +276,7 @@ func (t *timerQueueStandbyProcessorImpl) processWorkflowTimeout(timerTask *persi
 	return t.processTimer(timerTask, func(msBuilder *mutableStateBuilder) error {
 		// we do not need to notity new timer to base, since if there is no new event being replicated
 		// checking again if the timer can be completed is meaningless
-		return newTaskRetryError()
+		return ErrTaskRetry
 	})
 }
 

service/history/timerQueueStandbyProcessor_test.go

@@ -197,9 +197,7 @@ func (s *timerQueueStandbyProcessorSuite) TestProcessExpiredUserTimer_Pending()
 	persistenceMutableState := createMutableState(msBuilder)
 	s.mockExecutionMgr.On("GetWorkflowExecution", mock.Anything).Return(&persistence.GetWorkflowExecutionResponse{State: persistenceMutableState}, nil)
 
-	err := s.timerQueueStandbyProcessor.process(timerTask)
-	_, ok := err.(*taskRetryError)
-	s.True(ok)
+	s.Equal(ErrTaskRetry, s.timerQueueStandbyProcessor.process(timerTask))
 }
 
 func (s *timerQueueStandbyProcessorSuite) TestProcessExpiredUserTimer_Success() {
@@ -364,9 +362,7 @@ func (s *timerQueueStandbyProcessorSuite) TestProcessActivityTimeout_Pending() {
 	persistenceMutableState := createMutableState(msBuilder)
 	s.mockExecutionMgr.On("GetWorkflowExecution", mock.Anything).Return(&persistence.GetWorkflowExecutionResponse{State: persistenceMutableState}, nil).Once()
 
-	err := s.timerQueueStandbyProcessor.process(timerTask)
-	_, ok := err.(*taskRetryError)
-	s.True(ok)
+	s.Equal(ErrTaskRetry, s.timerQueueStandbyProcessor.process(timerTask))
 }
 
 func (s *timerQueueStandbyProcessorSuite) TestProcessActivityTimeout_Success() {
@@ -533,9 +529,7 @@ func (s *timerQueueStandbyProcessorSuite) TestProcessDecisionTimeout_Pending() {
 	persistenceMutableState := createMutableState(msBuilder)
 	s.mockExecutionMgr.On("GetWorkflowExecution", mock.Anything).Return(&persistence.GetWorkflowExecutionResponse{State: persistenceMutableState}, nil).Once()
 
-	err := s.timerQueueStandbyProcessor.process(timerTask)
-	_, ok := err.(*taskRetryError)
-	s.True(ok)
+	s.Equal(ErrTaskRetry, s.timerQueueStandbyProcessor.process(timerTask))
 }
 
 func (s *timerQueueStandbyProcessorSuite) TestProcessDecisionTimeout_Success() {
@@ -623,9 +617,7 @@ func (s *timerQueueStandbyProcessorSuite) TestProcessWorkflowTimeout_Pending() {
 	persistenceMutableState := createMutableState(msBuilder)
 	s.mockExecutionMgr.On("GetWorkflowExecution", mock.Anything).Return(&persistence.GetWorkflowExecutionResponse{State: persistenceMutableState}, nil).Once()
 
-	err := s.timerQueueStandbyProcessor.process(timerTask)
-	_, ok := err.(*taskRetryError)
-	s.True(ok)
+	s.Equal(ErrTaskRetry, s.timerQueueStandbyProcessor.process(timerTask))
 }
 
 func (s *timerQueueStandbyProcessorSuite) TestProcessWorkflowTimeout_Success() {