read autoconfighint from empty poll

Author: shijiesheng

internal/internal_task_handlers.go

@@ -75,15 +75,17 @@ type (
 	// workflowTask wraps a decision task.
 	workflowTask struct {
 		task            *s.PollForDecisionTaskResponse
+		autoConfigHint  *s.AutoConfigHint
 		historyIterator HistoryIterator
 		doneCh          chan struct{}
 		laResultCh      chan *localActivityResult
 	}
 
 	// activityTask wraps a activity task.
 	activityTask struct {
-		task          *s.PollForActivityTaskResponse
-		pollStartTime time.Time
+		task           *s.PollForActivityTaskResponse
+		autoConfigHint *s.AutoConfigHint
+		pollStartTime  time.Time
 	}
 
 	// resetStickinessTask wraps a ResetStickyTaskListRequest.

internal/internal_task_pollers.go

@@ -848,7 +848,9 @@ 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{}, nil
+		return &workflowTask{
+			autoConfigHint: response.GetAutoConfigHint(),
+		}, nil
 	}
 
 	wtp.updateBacklog(request.TaskList.GetKind(), response.GetBacklogCountHint())
@@ -908,6 +910,7 @@ func (wtp *workflowTaskPoller) toWorkflowTask(response *s.PollForDecisionTaskRes
 	task := &workflowTask{
 		task:            response,
 		historyIterator: historyIterator,
+		autoConfigHint:  response.GetAutoConfigHint(),
 	}
 	return task
 }
@@ -1116,7 +1119,9 @@ func (atp *activityTaskPoller) pollWithMetrics(ctx context.Context,
 		return nil, err
 	}
 	if response == nil || len(response.TaskToken) == 0 {
-		return &activityTask{}, nil
+		return &activityTask{
+			autoConfigHint: response.GetAutoConfigHint(),
+		}, nil
 	}
 
 	workflowType := response.WorkflowType.GetName()
@@ -1128,7 +1133,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}, nil
+	return &activityTask{task: response, pollStartTime: startTime, autoConfigHint: response.GetAutoConfigHint()}, nil
 }
 
 // PollTask polls a new task

internal/internal_worker_base.go

@@ -335,10 +335,11 @@ func (bw *baseWorker) pollTask() {
 		}
 	}
 
+	if bw.concurrencyAutoScaler != nil {
+		bw.concurrencyAutoScaler.ProcessPollerHint(getAutoConfigHint(task))
+	}
+
 	if task != nil {
-		if bw.concurrencyAutoScaler != nil {
-			bw.concurrencyAutoScaler.ProcessPollerHint(getAutoConfigHint(task))
-		}
 		select {
 		case bw.taskQueueCh <- &polledTask{task}:
 		case <-bw.shutdownCh:
@@ -428,15 +429,15 @@ func (bw *baseWorker) Stop() {
 func getAutoConfigHint(task interface{}) *shared.AutoConfigHint {
 	switch t := task.(type) {
 	case *workflowTask:
-		if t.task == nil {
+		if t.autoConfigHint == nil {
 			return nil
 		}
-		return t.task.AutoConfigHint
+		return t.autoConfigHint
 	case *activityTask:
-		if t.task == nil {
+		if t.autoConfigHint == nil {
 			return nil
 		}
-		return t.task.AutoConfigHint
+		return t.autoConfigHint
 	default:
 		return nil
 	}

internal/internal_worker_test.go

@@ -1514,13 +1514,25 @@ func TestGetTaskAutoConfigHint(t *testing.T) {
 		{
 			"decision task",
 			&workflowTask{
-				&shared.PollForDecisionTaskResponse{AutoConfigHint: &hint}, nil, nil, nil},
+				&shared.PollForDecisionTaskResponse{AutoConfigHint: &hint}, &hint, nil, nil, nil},
+			&hint,
+		},
+		{
+			"empty decision task",
+			&workflowTask{
+				nil, &hint, nil, nil, nil},
 			&hint,
 		},
 		{
 			"activity task",
 			&activityTask{
-				&shared.PollForActivityTaskResponse{AutoConfigHint: &hint}, time.Now()},
+				&shared.PollForActivityTaskResponse{AutoConfigHint: &hint}, &hint, time.Now()},
+			&hint,
+		},
+		{
+			"empty activity task",
+			&activityTask{
+				nil, &hint, time.Now()},
 			&hint,
 		},
 		{