populate task for empty polls instead

cadence-workflow · Jan 16, 2025 · d675887 · d675887
1 parent 9783bf3
commit d675887
Show file tree

Hide file tree

Showing 5 changed files with 16 additions and 21 deletions.
diff --git a/internal/internal_poller_autoscaler.go b/internal/internal_poller_autoscaler.go
@@ -174,9 +174,9 @@ func (m *pollerUsageEstimator) CollectUsage(data interface{}) error {
 func isTaskEmpty(task interface{}) (bool, error) {
 	switch t := task.(type) {
 	case *workflowTask:
-		return t == nil || t.task == nil, nil
+		return t == nil || isEmptyDecisionTask(t.task), nil
 	case *activityTask:
-		return t == nil || t.task == nil, nil
+		return t == nil || isEmptyActivityTask(t.task), nil
 	case *localActivityTask:
 		return t == nil || t.workflowTask == nil, nil
 	default:

diff --git a/internal/internal_task_handlers.go b/internal/internal_task_handlers.go
@@ -75,7 +75,6 @@ type (
 	// workflowTask wraps a decision task.
 	workflowTask struct {
 		task            *s.PollForDecisionTaskResponse
-		autoConfigHint  *s.AutoConfigHint
 		historyIterator HistoryIterator
 		doneCh          chan struct{}
 		laResultCh      chan *localActivityResult
@@ -84,7 +83,6 @@ type (
 	// activityTask wraps a activity task.
 	activityTask struct {
 		task           *s.PollForActivityTaskResponse
-		autoConfigHint *s.AutoConfigHint
 		pollStartTime  time.Time
 	}
 
@@ -746,7 +744,7 @@ func (wth *workflowTaskHandlerImpl) ProcessWorkflowTask(
 	workflowTask *workflowTask,
 	heartbeatFunc decisionHeartbeatFunc,
 ) (completeRequest interface{}, errRet error) {
-	if workflowTask == nil || workflowTask.task == nil {
+	if workflowTask == nil || isEmptyDecisionTask(workflowTask.task) {
 		return nil, errors.New("nil workflow task provided")
 	}
 	task := workflowTask.task

diff --git a/internal/internal_task_pollers.go b/internal/internal_task_pollers.go
@@ -848,9 +848,7 @@ func (wtp *workflowTaskPoller) poll(ctx context.Context) (interface{}, error) {
 	if response == nil || len(response.TaskToken) == 0 {
 		wtp.metricsScope.Counter(metrics.DecisionPollNoTaskCounter).Inc(1)
 		wtp.updateBacklog(request.TaskList.GetKind(), 0)
-		return &workflowTask{
-			autoConfigHint: response.GetAutoConfigHint(),
-		}, nil
+		return &workflowTask{task: response}, nil
 	}
 
 	wtp.updateBacklog(request.TaskList.GetKind(), response.GetBacklogCountHint())
@@ -910,7 +908,6 @@ func (wtp *workflowTaskPoller) toWorkflowTask(response *s.PollForDecisionTaskRes
 	task := &workflowTask{
 		task:            response,
 		historyIterator: historyIterator,
-		autoConfigHint:  response.GetAutoConfigHint(),
 	}
 	return task
 }
@@ -1119,9 +1116,7 @@ func (atp *activityTaskPoller) pollWithMetrics(ctx context.Context,
 		return nil, err
 	}
 	if response == nil || len(response.TaskToken) == 0 {
-		return &activityTask{
-			autoConfigHint: response.GetAutoConfigHint(),
-		}, nil
+		return &activityTask{task: response}, nil
 	}
 
 	workflowType := response.WorkflowType.GetName()
@@ -1133,7 +1128,7 @@ func (atp *activityTaskPoller) pollWithMetrics(ctx context.Context,
 	scheduledToStartLatency := time.Duration(response.GetStartedTimestamp() - response.GetScheduledTimestampOfThisAttempt())
 	metricsScope.Timer(metrics.ActivityScheduledToStartLatency).Record(scheduledToStartLatency)
 
-	return &activityTask{task: response, pollStartTime: startTime, autoConfigHint: response.GetAutoConfigHint()}, nil
+	return &activityTask{task: response, pollStartTime: startTime}, nil
 }
 
 // PollTask polls a new task

diff --git a/internal/internal_task_pollers_test.go b/internal/internal_task_pollers_test.go
@@ -87,15 +87,13 @@ func TestWorkflowTaskPoller(t *testing.T) {
 				task,
 				&workflowTask{
 					task: task,
-					autoConfigHint: task.AutoConfigHint,
 				},
 			},
 			{
 				"success with empty task",
 				emptyTask,
 				&workflowTask{
-					task: nil,
-					autoConfigHint: task.AutoConfigHint,
+					task: emptyTask,
 				},
 			},
 		} {
@@ -107,7 +105,6 @@ func TestWorkflowTaskPoller(t *testing.T) {
 				resultTask, ok := result.(*workflowTask)
 				assert.True(t, ok)
 				assert.Equal(t, tt.expected.task, resultTask.task)
-				assert.Equal(t, tt.expected.autoConfigHint, resultTask.autoConfigHint)
 			})
 		}
 	})
@@ -139,15 +136,13 @@ func TestActivityTaskPoller(t *testing.T) {
 				task,
 				&activityTask{
 					task: task,
-					autoConfigHint: task.AutoConfigHint,
 				},
 			},
 			{
 				"success with empty task",
 				emptyTask,
 				&activityTask{
-					task: nil,
-					autoConfigHint: task.AutoConfigHint,
+					task: emptyTask,
 				},
 			},
 		} {
@@ -159,7 +154,6 @@ func TestActivityTaskPoller(t *testing.T) {
 				resultTask, ok := result.(*activityTask)
 				assert.True(t, ok)
 				assert.Equal(t, tt.expected.task, resultTask.task)
-				assert.Equal(t, tt.expected.autoConfigHint, resultTask.autoConfigHint)
 			})
 		}
 	})

diff --git a/internal/internal_utils.go b/internal/internal_utils.go
@@ -492,3 +492,11 @@ func getLengthOfStringPointer(s *string) int {
 	}
 	return len(*s)
 }
+
+func isEmptyDecisionTask(r *s.PollForDecisionTaskResponse) bool {
+	return r == nil || len(r.TaskToken) == 0
+}
+
+func isEmptyActivityTask(r *s.PollForActivityTaskResponse) bool {
+	return r == nil || len(r.TaskToken) == 0
+}