Complete Transfer tasks even after execution is completed (#230)

Issue #229

Author: Tamer Eldeeb

service/history/transferQueueProcessor.go

@@ -305,6 +305,10 @@ func (t *transferQueueProcessorImpl) processActivityTask(task *persistence.Trans
 	timeout := int32(0)
 	if err != nil {
 		release()
+		if _, ok := err.(*workflow.EntityNotExistsError); ok {
+			// this could happen if this is a duplicate processing of the task, and the execution has already completed.
+			return nil
+		}
 		return err
 	}
 
@@ -339,6 +343,10 @@ func (t *transferQueueProcessorImpl) processDecisionTask(task *persistence.Trans
 	}
 
 	if err != nil {
+		if _, ok := err.(*workflow.EntityNotExistsError); ok {
+			// this could happen if this is a duplicate processing of the task, and the execution has already completed.
+			return nil
+		}
 		return err
 	}
 
@@ -454,6 +462,10 @@ func (t *transferQueueProcessorImpl) processCancelExecution(task *persistence.Tr
 	// Load workflow execution.
 	_, err = context.loadWorkflowExecution()
 	if err != nil {
+		if _, ok := err.(*workflow.EntityNotExistsError); ok {
+			// this could happen if this is a duplicate processing of the task, and the execution has already completed.
+			return nil
+		}
 		return err
 	}
 
@@ -500,6 +512,10 @@ func (t *transferQueueProcessorImpl) processStartChildExecution(task *persistenc
 	var msBuilder *mutableStateBuilder
 	msBuilder, err = context.loadWorkflowExecution()
 	if err != nil {
+		if _, ok := err.(*workflow.EntityNotExistsError); ok {
+			// this could happen if this is a duplicate processing of the task, and the execution has already completed.
+			return nil
+		}
 		return err
 	}
 

service/history/transferQueueProcessor_test.go

@@ -386,6 +386,33 @@ workerPump:
 	s.mockHistoryClient.AssertExpectations(s.T())
 }
 
+func (s *transferQueueProcessorSuite) TestCompleteTaskAfterExecutionDeleted() {
+	domainID := "b677a307-8261-40ea-b239-ab2ec78e443b"
+	workflowExecution := workflow.WorkflowExecution{WorkflowId: common.StringPtr("complete-task-execution-deleted-test"),
+		RunId: common.StringPtr("0d00698f-08e1-4d36-a3e2-3bf109f5d2d6")}
+	taskList := "complete-task-execution-deleted-queue"
+	task0, err0 := s.CreateWorkflowExecution(domainID, workflowExecution, taskList, "wType", 10, nil, 3, 0, 2, nil)
+	s.Nil(err0, "No error expected.")
+	s.NotEmpty(task0, "Expected non empty task identifier.")
+
+	tasksCh := make(chan *persistence.TransferTaskInfo, 10)
+	s.processor.processTransferTasks(tasksCh)
+workerPump:
+	for {
+		select {
+		case task := <-tasksCh:
+			if task.ScheduleID == firstEventID+1 {
+				s.mockVisibilityMgr.On("RecordWorkflowExecutionStarted", mock.Anything).Once().Return(&workflow.EntityNotExistsError{})
+			}
+			s.processor.processTransferTask(task)
+		default:
+			break workerPump
+		}
+	}
+
+	s.mockVisibilityMgr.AssertExpectations(s.T())
+}
+
 func createAddRequestFromTask(task *persistence.TransferTaskInfo, scheduleToStartTimeout int32) interface{} {
 	var res interface{}
 	domainID := task.DomainID