add target

common/persistence/serialization/task_serializer.go

@@ -1198,6 +1198,7 @@
 		Version:        syncWorkflowStateTask.Version,
 		VisibilityTime: timestamppb.New(syncWorkflowStateTask.VisibilityTimestamp),
 		Priority:       syncWorkflowStateTask.Priority,
+		// target Cluster id: syncWorkflowStateTask.TargetCluster,
 	}
 }
 
@@ -1218,6 +1219,7 @@
 		Version:             syncWorkflowStateTask.Version,
 		TaskID:              syncWorkflowStateTask.TaskId,
 		Priority:            syncWorkflowStateTask.Priority,
+		TargetCluster:    syncWorkflowStateTask.targetClusterId
 	}
 }
 

proto/internal/temporal/server/api/persistence/v1/executions.proto

@@ -380,6 +380,7 @@ message ReplicationTaskInfo {
     repeated ReplicationTaskInfo task_equivalents = 20;
     history.v1.VersionHistoryItem last_version_history_item = 21;
     bool is_first_task = 22;
+    string target_cluster_id = 23;
 }
 
 // visibility_task_data column

service/history/api/replication/generate_task.go

@@ -65,7 +65,7 @@ func GenerateTask(
 	defer func() { workflowLease.GetReleaseFn()(retError) }()
 
 	mutableState := workflowLease.GetMutableState()
-	replicationTasks, stateTransitionCount, err := mutableState.GenerateMigrationTasks()
+	replicationTasks, stateTransitionCount, err := mutableState.GenerateMigrationTasks("abc")
 	if err != nil {
 		return nil, err
 	}

service/history/interfaces/mutable_state.go

@@ -323,7 +323,7 @@ type (
 		// CloseTransactionAsSnapshot closes the mutable state transaction (different from DB transaction) and prepares the current snapshot of the state to be persisted and bumps the DBRecordVersion.
 		// You should ideally not make any changes to the mutable state after this call.
 		CloseTransactionAsSnapshot(transactionPolicy TransactionPolicy) (*persistence.WorkflowSnapshot, []*persistence.WorkflowEvents, error)
-		GenerateMigrationTasks() ([]tasks.Task, int64, error)
+		GenerateMigrationTasks(targetClusterId string) ([]tasks.Task, int64, error)
 
 		// ContinueAsNewMinBackoff calculate minimal backoff for next ContinueAsNew run.
 		// Input backoffDuration is current backoff for next run.

service/history/replication/stream_sender.go

@@ -635,6 +635,11 @@ func (s *StreamSenderImpl) shouldProcessTask(item tasks.Task) bool {
 		return false
 	}
 
+	targetCluster := s.getTaskTargetCluster(item)
+	if targetCluster != "" && targetCluster != s.clientClusterName {
+		return false
+	}
+
 	var shouldProcessTask bool
 	namespaceEntry, err := s.shardContext.GetNamespaceRegistry().GetNamespaceByID(
 		namespace.ID(item.GetNamespaceID()),
@@ -668,3 +673,12 @@ func (s *StreamSenderImpl) getTaskPriority(task tasks.Task) enumsspb.TaskPriorit
 		return enumsspb.TASK_PRIORITY_HIGH
 	}
 }
+
+func (s *StreamSenderImpl) getTaskTargetCluster(task tasks.Task) string {
+	switch t := task.(type) {
+	case *tasks.SyncWorkflowStateTask:
+		return t.TargetCluster
+	default:
+		return ""
+	}
+}

service/history/tasks/sync_workflow_state_task.go

@@ -39,8 +39,9 @@ type (
 		VisibilityTimestamp time.Time
 		TaskID              int64
 		// TODO: validate this version in source task converter
-		Version  int64
-		Priority enumsspb.TaskPriority
+		Version       int64
+		Priority      enumsspb.TaskPriority
+		TargetCluster string
 	}
 )
 

service/history/workflow/mutable_state_impl.go

@@ -5944,7 +5944,7 @@ func (ms *MutableStateImpl) UpdateDuplicatedResource(
 	ms.appliedEvents[id] = struct{}{}
 }
 
-func (ms *MutableStateImpl) GenerateMigrationTasks() ([]tasks.Task, int64, error) {
+func (ms *MutableStateImpl) GenerateMigrationTasks(targetClusterId string) ([]tasks.Task, int64, error) {
 	return ms.taskGenerator.GenerateMigrationTasks()
 }
 

service/history/workflow/task_generator.go

@@ -99,7 +99,7 @@ type (
 		GenerateHistoryReplicationTasks(
 			eventBatches [][]*historypb.HistoryEvent,
 		) ([]tasks.Task, error)
-		GenerateMigrationTasks() ([]tasks.Task, int64, error)
+		GenerateMigrationTasks(targetClusterId string) ([]tasks.Task, int64, error)
 
 		// Generate tasks for any updated state machines on mutable state.
 		// Looks up machine definition in the provided registry.
@@ -739,7 +739,7 @@ func (r *TaskGeneratorImpl) GenerateHistoryReplicationTasks(
 	}, nil
 }
 
-func (r *TaskGeneratorImpl) GenerateMigrationTasks() ([]tasks.Task, int64, error) {
+func (r *TaskGeneratorImpl) GenerateMigrationTasks(targetClusterId string) ([]tasks.Task, int64, error) {
 	executionInfo := r.mutableState.GetExecutionInfo()
 	versionHistory, err := versionhistory.GetCurrentVersionHistory(executionInfo.GetVersionHistories())
 	if err != nil {
@@ -755,9 +755,10 @@ func (r *TaskGeneratorImpl) GenerateMigrationTasks() ([]tasks.Task, int64, error
 	if r.mutableState.GetExecutionState().State == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED {
 		syncWorkflowStateTask := []tasks.Task{&tasks.SyncWorkflowStateTask{
 			// TaskID, VisibilityTimestamp is set by shard
-			WorkflowKey: workflowKey,
-			Version:     lastItem.GetVersion(),
-			Priority:    enumsspb.TASK_PRIORITY_LOW,
+			WorkflowKey:   workflowKey,
+			Version:       lastItem.GetVersion(),
+			Priority:      enumsspb.TASK_PRIORITY_LOW,
+			TargetCluster: targetClusterId,
 		}}
 		if r.mutableState.IsTransitionHistoryEnabled() &&
 			// even though current cluster may enabled state transition, but transition history can be cleared