Next retry delay should not be bounded by retry policy MaximumInterval

service/history/workflow/activity.go

@@ -75,14 +75,20 @@ func newActivityVisitor(
 	}
 
 	now := timesource.Now().In(time.UTC)
+	retryMaxInterval := ai.RetryMaximumInterval
 	delay := nextRetryDelayFrom(failure)
 
+	// if a delay is specified by the application it should override the maximum interval set by the retry policy.
+	if delay != nil {
+		retryMaxInterval = durationpb.New(*delay)
+	}
+
 	backoff, retryState := nextBackoffInterval(
 		now,
 		ai.Attempt,
 		ai.RetryMaximumAttempts,
 		ai.RetryInitialInterval,
-		ai.RetryMaximumInterval,
+		retryMaxInterval,
 		ai.RetryExpirationTime,
 		ai.RetryBackoffCoefficient,
 		makeBackoffAlgorithm(delay),

service/history/workflow/mutable_state_impl_restart_activity_test.go

@@ -208,6 +208,24 @@ func (s *retryActivitySuite) TestRetryActivity_should_be_scheduled_when_next_ret
 	s.assertTruncateFailureCalled()
 }
 
+func (s *retryActivitySuite) TestRetryActivity_next_retry_delay_should_override_max_interval() {
+	s.mutableState.timeSource = s.timeSource
+	taskGeneratorMock := NewMockTaskGenerator(s.controller)
+	nextAttempt := s.activity.Attempt + 1
+	expectedScheduledTime := s.timeSource.Now().Add(3 * time.Minute).UTC()
+	taskGeneratorMock.EXPECT().GenerateActivityRetryTasks(s.activity.ScheduledEventId, expectedScheduledTime, nextAttempt)
+	s.mutableState.taskGenerator = taskGeneratorMock
+
+	s.failure.GetApplicationFailureInfo().NextRetryDelay = durationpb.New(3 * time.Minute)
+	s.activity.RetryMaximumInterval = durationpb.New(2 * time.Minute) // set retry max interval to be less than next retry delay duration.
+	_, err := s.mutableState.RetryActivity(s.activity, s.failure)
+	s.NoError(err)
+	s.Equal(s.activity.Attempt, nextAttempt)
+
+	s.Equal(expectedScheduledTime, s.activity.ScheduledTime.AsTime(), "Activity scheduled time is incorrect")
+	s.assertTruncateFailureCalled()
+}
+
 func (s *retryActivitySuite) TestRetryActivity_when_no_next_backoff_interval_should_fail() {
 	taskGeneratorMock := NewMockTaskGenerator(s.controller)
 	s.mutableState.taskGenerator = taskGeneratorMock