Fix replication task serialization (#6741)

Author: Shaddoll

Diff for: service/history/execution/mutable_state_builder.go

@@ -1772,6 +1772,11 @@ func (e *mutableStateBuilder) eventsToReplicationTask(
 
 	// the visibility timestamp will be set in shard context
 	replicationTask := &persistence.HistoryReplicationTask{
+		WorkflowIdentifier: persistence.WorkflowIdentifier{
+			DomainID:   e.executionInfo.DomainID,
+			WorkflowID: e.executionInfo.WorkflowID,
+			RunID:      e.executionInfo.RunID,
+		},
 		TaskData: persistence.TaskData{
 			Version: firstEvent.Version,
 		},
@@ -1794,6 +1799,7 @@ func (e *mutableStateBuilder) syncActivityToReplicationTask(
 	}
 
 	return convertSyncActivityInfos(
+		e.executionInfo,
 		e.pendingActivityInfoIDs,
 		e.syncActivityTasks,
 	)

Diff for: service/history/execution/mutable_state_builder_test.go

@@ -3536,6 +3536,11 @@ func TestCloseTransactionAsMutation(t *testing.T) {
 					persistence.HistoryTaskCategoryTimer:    nil,
 					persistence.HistoryTaskCategoryReplication: []persistence.Task{
 						&persistence.HistoryReplicationTask{
+							WorkflowIdentifier: persistence.WorkflowIdentifier{
+								DomainID:   "some-domain-id",
+								WorkflowID: "",
+								RunID:      "",
+							},
 							FirstEventID: 1,
 							NextEventID:  2,
 							TaskData: persistence.TaskData{

Diff for: service/history/execution/mutable_state_util.go

@@ -46,6 +46,7 @@ func (policy TransactionPolicy) Ptr() *TransactionPolicy {
 }
 
 func convertSyncActivityInfos(
+	executionInfo *persistence.WorkflowExecutionInfo,
 	activityInfos map[int64]*persistence.ActivityInfo,
 	inputs map[int64]struct{},
 ) []persistence.Task {
@@ -55,6 +56,11 @@ func convertSyncActivityInfos(
 		if ok {
 			// the visibility timestamp will be set in shard context
 			outputs = append(outputs, &persistence.SyncActivityTask{
+				WorkflowIdentifier: persistence.WorkflowIdentifier{
+					DomainID:   executionInfo.DomainID,
+					WorkflowID: executionInfo.WorkflowID,
+					RunID:      executionInfo.RunID,
+				},
 				TaskData: persistence.TaskData{
 					Version: activityInfo.Version,
 				},

Diff for: service/history/execution/mutable_state_util_test.go

@@ -457,13 +457,21 @@ func TestGetChildExecutionDomainEntry(t *testing.T) {
 
 func TestConvert(t *testing.T) {
 	t.Run("convertSyncActivityInfos", func(t *testing.T) {
+		executionInfo := &persistence.WorkflowExecutionInfo{
+			DomainID:   "some-domain-id",
+			WorkflowID: "some-workflow-id",
+			RunID:      "some-run-id",
+		}
 		activityInfos := map[int64]*persistence.ActivityInfo{1: {Version: 1, ScheduleID: 1}}
 		inputs := map[int64]struct{}{1: {}}
-		outputs := convertSyncActivityInfos(activityInfos, inputs)
+		outputs := convertSyncActivityInfos(executionInfo, activityInfos, inputs)
 		assert.NotNil(t, outputs)
 		assert.Equal(t, 1, len(outputs))
 		assert.Equal(t, int64(1), outputs[0].(*persistence.SyncActivityTask).ScheduledID)
 		assert.Equal(t, int64(1), outputs[0].GetVersion())
+		assert.Equal(t, executionInfo.DomainID, outputs[0].(*persistence.SyncActivityTask).DomainID)
+		assert.Equal(t, executionInfo.WorkflowID, outputs[0].(*persistence.SyncActivityTask).WorkflowID)
+		assert.Equal(t, executionInfo.RunID, outputs[0].(*persistence.SyncActivityTask).RunID)
 	})
 }