Consolidate and minor fix on retry behavior for GetHistoryTasks operation (#6957)

Author: timl3136

service/history/queue/timer_queue_processor_base.go

@@ -48,6 +48,7 @@ var (
 		time.Unix(0, math.MaxInt64),
 		0,
 	)
+	timerTaskOperationRetryPolicy = common.CreatePersistenceRetryPolicy()
 )
 
 type (
@@ -518,19 +519,24 @@ func (t *timerQueueProcessorBase) getTimerTasks(readLevel, maxReadLevel task.Key
 		PageSize:            batchSize,
 		NextPageToken:       nextPageToken,
 	}
-	var err error
+
 	var response *persistence.GetHistoryTasksResponse
-	retryCount := t.shard.GetConfig().TimerProcessorGetFailureRetryCount()
-	for attempt := 0; attempt < retryCount; attempt++ {
-		response, err = t.shard.GetExecutionManager().GetHistoryTasks(context.Background(), request)
-		if err == nil {
-			return response, nil
-		}
-		backoff := time.Duration(attempt*100) * time.Millisecond
-		t.logger.Debugf("Failed to get timer tasks from execution manager. error: %v, attempt: %d, retryCount: %d, backoff: %v", err, attempt, retryCount, backoff)
-		time.Sleep(backoff)
+	op := func(ctx context.Context) error {
+		var err error
+		response, err = t.shard.GetExecutionManager().GetHistoryTasks(ctx, request)
+		return err
+	}
+
+	throttleRetry := backoff.NewThrottleRetry(
+		backoff.WithRetryPolicy(timerTaskOperationRetryPolicy),
+		backoff.WithRetryableError(func(err error) bool { return true }),
+	)
+	err := throttleRetry.Do(context.Background(), op)
+	if err != nil {
+		return nil, err
 	}
-	return nil, err
+
+	return response, nil
 }
 
 func (t *timerQueueProcessorBase) isProcessNow(expiryTime time.Time) bool {

service/history/queue/timer_queue_processor_base_test.go

@@ -495,7 +495,7 @@ func (s *timerQueueProcessorBaseSuite) TestReadAndFilterTasks_LookAheadFailed_No
 
 	mockExecutionMgr := s.mockShard.Resource.ExecutionMgr
 	mockExecutionMgr.On("GetHistoryTasks", mock.Anything, request).Return(response, nil).Once()
-	mockExecutionMgr.On("GetHistoryTasks", mock.Anything, lookAheadRequest).Return(nil, errors.New("some random error")).Times(s.mockShard.GetConfig().TimerProcessorGetFailureRetryCount())
+	mockExecutionMgr.On("GetHistoryTasks", mock.Anything, lookAheadRequest).Return(nil, errors.New("some random error")).Times(0)
 
 	timerQueueProcessBase, done := s.newTestTimerQueueProcessorBase(nil, nil, nil, nil, nil)
 	defer done()

service/history/queue/transfer_queue_processor_base.go

@@ -528,7 +528,7 @@ func (t *transferQueueProcessorBase) readTasks(
 	var response *persistence.GetHistoryTasksResponse
 	op := func(ctx context.Context) error {
 		var err error
-		response, err = t.shard.GetExecutionManager().GetHistoryTasks(context.Background(), &persistence.GetHistoryTasksRequest{
+		response, err = t.shard.GetExecutionManager().GetHistoryTasks(ctx, &persistence.GetHistoryTasksRequest{
 			TaskCategory:        persistence.HistoryTaskCategoryTransfer,
 			InclusiveMinTaskKey: persistence.NewImmediateTaskKey(readLevel.(transferTaskKey).taskID + 1),
 			ExclusiveMaxTaskKey: persistence.NewImmediateTaskKey(maxReadLevel.(transferTaskKey).taskID + 1),

service/history/queue/transfer_queue_processor_test.go

@@ -208,7 +208,7 @@ func TestTransferQueueProcessor_FailoverDomain(t *testing.T) {
 						},
 					},
 				}
-				mockShard.GetExecutionManager().(*mocks.ExecutionManager).On("GetHistoryTasks", context.Background(), mock.Anything).Return(response, nil).Once()
+				mockShard.GetExecutionManager().(*mocks.ExecutionManager).On("GetHistoryTasks", mock.Anything, mock.Anything).Return(response, nil).Once()
 			},
 			processorStarted: true,
 		},