Get execution start/close time when using cassandra visibility store (#4549)

Author: rodrigozhou

service/history/deletemanager/delete_manager.go

@@ -40,6 +40,7 @@ import (
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/persistence"
 	"go.temporal.io/server/common/persistence/visibility/manager"
+	"go.temporal.io/server/common/persistence/visibility/store/standard/cassandra"
 	"go.temporal.io/server/common/primitives"
 	"go.temporal.io/server/common/searchattribute"
 	"go.temporal.io/server/service/history/configs"
@@ -210,15 +211,17 @@ func (m *DeleteManagerImpl) deleteWorkflowExecutionInternal(
 	// TODO (alex): Remove them when cassandra standard visibility is removed.
 	var startTime *time.Time
 	var closeTime *time.Time
-	// There are cases when workflow execution is closed but visibility is not updated and still open.
-	// This happens, for example, when workflow execution is deleted right from CloseExecutionTask.
-	// Therefore, force to delete from open visibility regardless of execution state.
-	if forceDeleteFromOpenVisibility || ms.GetExecutionState().State != enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED {
-		startTime = ms.GetExecutionInfo().GetStartTime()
-	} else {
-		closeTime, err = ms.GetWorkflowCloseTime(ctx)
-		if err != nil {
-			return err
+	if m.visibilityManager.HasStoreName(cassandra.CassandraPersistenceName) {
+		// There are cases when workflow execution is closed but visibility is not updated and still open.
+		// This happens, for example, when workflow execution is deleted right from CloseExecutionTask.
+		// Therefore, force to delete from open visibility regardless of execution state.
+		if forceDeleteFromOpenVisibility || ms.GetExecutionState().State != enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED {
+			startTime = ms.GetExecutionInfo().GetStartTime()
+		} else {
+			closeTime, err = ms.GetWorkflowCloseTime(ctx)
+			if err != nil {
+				return err
+			}
 		}
 	}
 

service/history/deletemanager/delete_manager_test.go

@@ -47,6 +47,7 @@ import (
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/persistence/visibility/manager"
+	"go.temporal.io/server/common/persistence/visibility/store/standard/cassandra"
 	"go.temporal.io/server/common/primitives"
 	"go.temporal.io/server/common/searchattribute"
 	"go.temporal.io/server/service/history/shard"
@@ -122,6 +123,7 @@ func (s *deleteManagerWorkflowSuite) TestDeleteDeletedWorkflowExecution() {
 		RunId:      tests.RunID,
 	}
 
+	s.mockVisibilityManager.EXPECT().HasStoreName(cassandra.CassandraPersistenceName).Return(true)
 	mockWeCtx := workflow.NewMockContext(s.controller)
 	mockMutableState := workflow.NewMockMutableState(s.controller)
 	mockMutableState.EXPECT().GetCurrentBranchToken().Return([]byte{22, 8, 78}, nil)
@@ -167,6 +169,7 @@ func (s *deleteManagerWorkflowSuite) TestDeleteDeletedWorkflowExecution_Error()
 		RunId:      tests.RunID,
 	}
 
+	s.mockVisibilityManager.EXPECT().HasStoreName(cassandra.CassandraPersistenceName).Return(true)
 	mockWeCtx := workflow.NewMockContext(s.controller)
 	mockMutableState := workflow.NewMockMutableState(s.controller)
 	mockMutableState.EXPECT().GetCurrentBranchToken().Return([]byte{22, 8, 78}, nil)
@@ -212,6 +215,7 @@ func (s *deleteManagerWorkflowSuite) TestDeleteWorkflowExecution_OpenWorkflow()
 	}
 	now := time.Now()
 
+	s.mockVisibilityManager.EXPECT().HasStoreName(cassandra.CassandraPersistenceName).Return(true)
 	mockWeCtx := workflow.NewMockContext(s.controller)
 	mockMutableState := workflow.NewMockMutableState(s.controller)
 	closeExecutionVisibilityTaskID := int64(39)
@@ -257,6 +261,7 @@ func (s *deleteManagerWorkflowSuite) TestDeleteWorkflowExecutionRetention_Archiv
 				RunId:      tests.RunID,
 			}
 
+			s.mockVisibilityManager.EXPECT().HasStoreName(cassandra.CassandraPersistenceName).Return(true)
 			mockWeCtx := workflow.NewMockContext(s.controller)
 			mockMutableState := workflow.NewMockMutableState(s.controller)
 
@@ -332,6 +337,7 @@ func (s *deleteManagerWorkflowSuite) TestDeleteWorkflowExecutionRetention_Archiv
 				RunId:      tests.RunID,
 			}
 
+			s.mockVisibilityManager.EXPECT().HasStoreName(cassandra.CassandraPersistenceName).Return(true)
 			mockWeCtx := workflow.NewMockContext(s.controller)
 			mockMutableState := workflow.NewMockMutableState(s.controller)
 			branchToken := []byte{22, 8, 78}