Add CloseVisibilityTaskID to DeleteExecutionVisibilityTask (#3391)

temporalio · Sep 15, 2022 · c91d596 · c91d596
1 parent ce9a66f
commit c91d596
Show file tree

Hide file tree

Showing 6 changed files with 306 additions and 235 deletions.
diff --git a/api/persistence/v1/executions.pb.go b/api/persistence/v1/executions.pb.go
diff --git a/common/persistence/serialization/task_serializer.go b/common/persistence/serialization/task_serializer.go
@@ -927,15 +927,16 @@ func (s *TaskSerializer) visibilityDeleteTaskToProto(
 	deleteVisibilityTask *tasks.DeleteExecutionVisibilityTask,
 ) *persistencespb.VisibilityTaskInfo {
 	return &persistencespb.VisibilityTaskInfo{
-		NamespaceId:    deleteVisibilityTask.WorkflowKey.NamespaceID,
-		WorkflowId:     deleteVisibilityTask.WorkflowKey.WorkflowID,
-		RunId:          deleteVisibilityTask.WorkflowKey.RunID,
-		TaskType:       enumsspb.TASK_TYPE_VISIBILITY_DELETE_EXECUTION,
-		Version:        deleteVisibilityTask.Version,
-		TaskId:         deleteVisibilityTask.TaskID,
-		VisibilityTime: &deleteVisibilityTask.VisibilityTimestamp,
-		StartTime:      deleteVisibilityTask.StartTime,
-		CloseTime:      deleteVisibilityTask.CloseTime,
+		NamespaceId:           deleteVisibilityTask.WorkflowKey.NamespaceID,
+		WorkflowId:            deleteVisibilityTask.WorkflowKey.WorkflowID,
+		RunId:                 deleteVisibilityTask.WorkflowKey.RunID,
+		TaskType:              enumsspb.TASK_TYPE_VISIBILITY_DELETE_EXECUTION,
+		Version:               deleteVisibilityTask.Version,
+		TaskId:                deleteVisibilityTask.TaskID,
+		VisibilityTime:        &deleteVisibilityTask.VisibilityTimestamp,
+		StartTime:             deleteVisibilityTask.StartTime,
+		CloseTime:             deleteVisibilityTask.CloseTime,
+		CloseVisibilityTaskId: deleteVisibilityTask.CloseVisibilityTaskID,
 	}
 }
 
@@ -948,11 +949,12 @@ func (s *TaskSerializer) visibilityDeleteTaskFromProto(
 			deleteVisibilityTask.WorkflowId,
 			deleteVisibilityTask.RunId,
 		),
-		VisibilityTimestamp: *deleteVisibilityTask.VisibilityTime,
-		TaskID:              deleteVisibilityTask.TaskId,
-		Version:             deleteVisibilityTask.Version,
-		StartTime:           deleteVisibilityTask.StartTime,
-		CloseTime:           deleteVisibilityTask.CloseTime,
+		VisibilityTimestamp:   *deleteVisibilityTask.VisibilityTime,
+		TaskID:                deleteVisibilityTask.TaskId,
+		Version:               deleteVisibilityTask.Version,
+		StartTime:             deleteVisibilityTask.StartTime,
+		CloseTime:             deleteVisibilityTask.CloseTime,
+		CloseVisibilityTaskID: deleteVisibilityTask.CloseVisibilityTaskId,
 	}
 }
 

diff --git a/common/persistence/serialization/task_serializer_test.go b/common/persistence/serialization/task_serializer_test.go
@@ -330,6 +330,29 @@ func (s *taskSerializerSuite) TestReplicateHistoryTask() {
 	s.assertEqualTasks(replicateHistoryTask)
 }
 
+func (s *taskSerializerSuite) TestDeleteExecutionVisibilityTask() {
+	replicateHistoryTask := &tasks.DeleteExecutionVisibilityTask{
+		WorkflowKey:           s.workflowKey,
+		VisibilityTimestamp:   time.Unix(0, 0).UTC(), // go == compare for location as well which is striped during marshaling/unmarshaling
+		TaskID:                rand.Int63(),
+		Version:               rand.Int63(),
+		CloseVisibilityTaskID: rand.Int63(),
+	}
+
+	s.assertEqualTasks(replicateHistoryTask)
+}
+
+func (s *taskSerializerSuite) TestDeleteExecutionTask() {
+	replicateHistoryTask := &tasks.DeleteExecutionTask{
+		WorkflowKey:         s.workflowKey,
+		VisibilityTimestamp: time.Unix(0, 0).UTC(), // go == compare for location as well which is striped during marshaling/unmarshaling
+		TaskID:              rand.Int63(),
+		Version:             rand.Int63(),
+	}
+
+	s.assertEqualTasks(replicateHistoryTask)
+}
+
 func (s *taskSerializerSuite) assertEqualTasks(
 	task tasks.Task,
 ) {

diff --git a/proto/internal/temporal/server/api/persistence/v1/executions.proto b/proto/internal/temporal/server/api/persistence/v1/executions.proto
@@ -61,7 +61,7 @@ message ShardInfo {
     reserved 15;
     // Map from task category to ack levels of the corresponding queue processor
     map<int32, QueueAckLevel> queue_ack_levels = 16;
-    map<int32, QueueState> queue_states = 17;  
+    map<int32, QueueState> queue_states = 17;
 }
 
 // execution column
@@ -205,6 +205,7 @@ message VisibilityTaskInfo {
     google.protobuf.Timestamp visibility_time = 7 [(gogoproto.stdtime) = true];
     google.protobuf.Timestamp close_time = 8 [(gogoproto.stdtime) = true];
     google.protobuf.Timestamp start_time = 9 [(gogoproto.stdtime) = true];
+    int64 close_visibility_task_id = 10;
 }
 
 // timer column

diff --git a/service/history/shard/controller_test.go b/service/history/shard/controller_test.go
@@ -163,7 +163,7 @@ func (s *controllerSuite) TestAcquireShardSuccess() {
 		if hostID == 0 {
 			myShards = append(myShards, shardID)
 			s.mockHistoryEngine.EXPECT().Start().Return()
-			s.mockHistoryEngine.EXPECT().NotifyNewTasks(gomock.Any(), gomock.Any())
+			s.mockHistoryEngine.EXPECT().NotifyNewTasks(gomock.Any(), gomock.Any()).AnyTimes()
 			s.mockServiceResolver.EXPECT().Lookup(convert.Int32ToString(shardID)).Return(s.hostInfo, nil).Times(2)
 			s.mockEngineFactory.EXPECT().CreateEngine(gomock.Any()).Return(s.mockHistoryEngine)
 			s.mockShardManager.EXPECT().GetOrCreateShard(gomock.Any(), getOrCreateShardRequestMatcher(shardID)).Return(

diff --git a/service/history/tasks/delete_visibility_task.go b/service/history/tasks/delete_visibility_task.go
@@ -36,9 +36,10 @@ var _ Task = (*DeleteExecutionVisibilityTask)(nil)
 type (
 	DeleteExecutionVisibilityTask struct {
 		definition.WorkflowKey
-		VisibilityTimestamp time.Time
-		TaskID              int64
-		Version             int64
+		VisibilityTimestamp   time.Time
+		TaskID                int64
+		Version               int64
+		CloseVisibilityTaskID int64
 		// These two fields are needed for cassandra standard visibility.
 		// TODO (alex): Remove them when cassandra standard visibility is removed.
 		StartTime *time.Time
@@ -81,3 +82,7 @@ func (t *DeleteExecutionVisibilityTask) GetCategory() Category {
 func (t *DeleteExecutionVisibilityTask) GetType() enumsspb.TaskType {
 	return enumsspb.TASK_TYPE_VISIBILITY_DELETE_EXECUTION
 }
+
+func (t *DeleteExecutionVisibilityTask) GetCloseVisibilityTaskID() int64 {
+	return t.CloseVisibilityTaskID
+}