303
task *replicationspb.ReplicationTask,
304
forceApply bool,
306
>
307
>
attr := task.GetSyncWorkflowStateTaskAttributes()
308
>
executionInfo := attr.GetWorkflowState().GetExecutionInfo()
309
>
namespaceID := namespace.ID(executionInfo.GetNamespaceId())
310
>
311
>
doContinue, err := e.filterTask(namespaceID, executionInfo.GetWorkflowId(), forceApply)
312
>
if err != nil || !doContinue {
313
return err
314
}
315
316
>
namespaceName, _ := e.namespaceRegistry.GetNamespaceName(namespace.ID(executionInfo.NamespaceId))
task_executor.go
317
>
ctx, cancel := e.newTaskContext(ctx, namespaceName)
318
>
defer cancel()
319
>
320
>
// This might be extra cost if the workflow belongs to local shard.
321
>
// Add a wrapper of the history client to call history engine directly if it becomes an issue.
322
>
request := &historyservice.ReplicateWorkflowStateRequest{
323
>
NamespaceId: namespaceID.String(),
324
>
WorkflowState: attr.GetWorkflowState(),
325
>
RemoteCluster: e.remoteCluster,
326
>
}
327
>
_, err = e.shardContext.GetHistoryClient().ReplicateWorkflowState(ctx, request)
328
>
switch retryErr := err.(type) {
329
>
case nil:
330
>
return nil
331
case *serviceerrors.RetryReplication:
332
resendErr := e.resend(