modified visibility store to support custom visibility store #6168
Conversation
common/persistence/visibility/store/elasticsearch/client/client.go
Outdated
Show resolved
Hide resolved
| CreateIndex(ctx context.Context, index string, body map[string]any) (bool, error) | ||
| DeleteIndex(ctx context.Context, indexName string) (bool, error) | ||
| IndexPutTemplate(ctx context.Context, templateName string, bodyString string) (bool, error) |
There was a problem hiding this comment.
Undo changes here. Actually, remove the ones you're adding to Client interface.
| wfCloseTime, err := mutableState.GetWorkflowCloseTime(ctx) | ||
| if err != nil { | ||
| return err | ||
| } |
There was a problem hiding this comment.
What this function returns if the workflow is still running? Don't need to check it first?
There was a problem hiding this comment.
I believe then the closeTime is nil, and it panics when it tries to convert nil to time. I added a check to only get closeTime when workflow is not running. Also, in GetWorkflowCloseTime, I added error handling for when closeTime is nil, just in case when processDeleteExecution is called for workflow with the status "zombie" or "corrupted" and closeTime is nil.
There was a problem hiding this comment.
removed this code. We are getting workflow closetime from task as mutable state may be deleted when processing delete execution.
| @@ -73,6 +75,7 @@ func NewManager( | |||
| metricsHandler metrics.Handler, | |||
| logger log.Logger, | |||
| ) (manager.VisibilityManager, error) { | |||
| fmt.Println("saas-temporal: test: new manager called") | |||
| if body == nil { | ||
| body = make(map[string]interface{}) | ||
| } |
There was a problem hiding this comment.
Is this really necessary? I think it's fine to call BodyJson(nil).
There was a problem hiding this comment.
I was getting this error, and I believe `BodyJson(nil).Do()` causes this issue.
triggered here at setupIndex for test cluster
| @@ -115,7 +115,7 @@ type ( | |||
| GetWorkflowExecution(ctx context.Context, request *persistence.GetWorkflowExecutionRequest) (*persistence.GetWorkflowExecutionResponse, error) | |||
| // DeleteWorkflowExecution add task to delete visibility, current workflow execution, and deletes workflow execution. | |||
| // If branchToken != nil, then delete history also, otherwise leave history. | |||
| DeleteWorkflowExecution(ctx context.Context, workflowKey definition.WorkflowKey, branchToken []byte, closeExecutionVisibilityTaskID int64, stage *tasks.DeleteWorkflowExecutionStage) error | |||
| DeleteWorkflowExecution(ctx context.Context, workflowKey definition.WorkflowKey, branchToken []byte, closeExecutionVisibilityTaskID int64, workflowCloseTime time.Time, stage *tasks.DeleteWorkflowExecutionStage) error | |||
There was a problem hiding this comment.
There was a problem hiding this comment.
yea I added as time.Time because when we are creating visibility close task from proto, deleteVisibilityTask.CloseTime.AsTime() always return time.Time.
I also noticed that we are passing closeExecutionVisibilityTaskID as non-pointer object and we identify if it is "nil" by checking if that parameter equals to 0. So, I thought maybe we could do the similar thing for the closetime as well (i.e., identity if the closetime exists by checking if task.WorkflowCloseTime.After(time.Unix(0, 0).
| if ms.executionInfo.CloseTime == nil { | ||
| return time.Time{}, ErrMissingCloseTimeInfo | ||
| } |
There was a problem hiding this comment.
This is a breaking change. Let's not do this.
There was a problem hiding this comment.
ah that is right. It will return error if I have this code block. I removed them
What changed?
Why?
How did you test it?
Potential risks
Documentation
Is hotfix candidate?