Add task token to activityDispatchInfo for worker (cadence-workflow#3672

)
vytautas-karpavicius · Feb 4, 2021 · 98b9f59 · 98b9f59
1 parent 4787d1c
commit 98b9f59
Show file tree

Hide file tree

Showing 3 changed files with 22 additions and 0 deletions.
diff --git a/service/history/decisionHandler.go b/service/history/decisionHandler.go
@@ -417,6 +417,7 @@ Update_History_Loop:
 				msBuilder,
 				handler.decisionAttrValidator,
 				workflowSizeChecker,
+				handler.tokenSerializer,
 				handler.logger,
 				handler.domainCache,
 				handler.metricsClient,

diff --git a/service/history/decisionTaskHandler.go b/service/history/decisionTaskHandler.go
@@ -59,6 +59,8 @@ type (
 		attrValidator    *decisionAttrValidator
 		sizeLimitChecker *workflowSizeChecker
 
+		tokenSerializer common.TaskTokenSerializer
+
 		logger        log.Logger
 		domainCache   cache.DomainCache
 		metricsClient metrics.Client
@@ -77,6 +79,7 @@ func newDecisionTaskHandler(
 	mutableState execution.MutableState,
 	attrValidator *decisionAttrValidator,
 	sizeLimitChecker *workflowSizeChecker,
+	tokenSerializer common.TaskTokenSerializer,
 	logger log.Logger,
 	domainCache cache.DomainCache,
 	metricsClient metrics.Client,
@@ -102,6 +105,8 @@ func newDecisionTaskHandler(
 		attrValidator:    attrValidator,
 		sizeLimitChecker: sizeLimitChecker,
 
+		tokenSerializer: tokenSerializer,
+
 		logger:        logger,
 		domainCache:   domainCache,
 		metricsClient: metricsClient,
@@ -249,6 +254,20 @@ func (handler *decisionTaskHandlerImpl) handleDecisionScheduleActivity(
 			if _, err1 := handler.mutableState.AddActivityTaskStartedEvent(ai, event.GetEventId(), uuid.New(), handler.identity); err1 != nil {
 				return nil, err1
 			}
+			token := &common.TaskToken{
+				DomainID:        executionInfo.DomainID,
+				WorkflowID:      executionInfo.WorkflowID,
+				WorkflowType:    executionInfo.WorkflowTypeName,
+				RunID:           executionInfo.RunID,
+				ScheduleID:      ai.ScheduleID,
+				ScheduleAttempt: 0,
+				ActivityID:      ai.ActivityID,
+				ActivityType:    attr.ActivityType.GetName(),
+			}
+			activityDispatchInfo.TaskToken, err = handler.tokenSerializer.Serialize(token)
+			if err != nil {
+				return nil, ErrSerializingToken
+			}
 			activityDispatchInfo.ScheduledTimestamp = common.Int64Ptr(ai.ScheduledTime.UnixNano())
 			activityDispatchInfo.ScheduledTimestampOfThisAttempt = common.Int64Ptr(ai.ScheduledTime.UnixNano())
 			activityDispatchInfo.StartedTimestamp = common.Int64Ptr(ai.StartedTime.UnixNano())

diff --git a/service/history/historyEngine.go b/service/history/historyEngine.go
@@ -135,6 +135,8 @@ var (
 	ErrWorkflowParent = &workflow.EntityNotExistsError{Message: "workflow parent does not match"}
 	// ErrDeserializingToken is the error to indicate task token is invalid
 	ErrDeserializingToken = &workflow.BadRequestError{Message: "error deserializing task token"}
+	// ErrSerializingToken is the error to indicate task token can not be serialized
+	ErrSerializingToken = &workflow.BadRequestError{Message: "error serializing task token"}
 	// ErrSignalOverSize is the error to indicate signal input size is > 256K
 	ErrSignalOverSize = &workflow.BadRequestError{Message: "signal input size is over 256K"}
 	// ErrCancellationAlreadyRequested is the error indicating cancellation for target workflow is already requested