diff --git a/service/history/workflow/activity.go b/service/history/workflow/activity.go
index 40cbfd03189..2d504411ab8 100644
--- a/service/history/workflow/activity.go
+++ b/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),
diff --git a/service/history/workflow/mutable_state_impl_restart_activity_test.go b/service/history/workflow/mutable_state_impl_restart_activity_test.go
index 413f063b57f..2164b55f5e7 100644
--- a/service/history/workflow/mutable_state_impl_restart_activity_test.go
+++ b/service/history/workflow/mutable_state_impl_restart_activity_test.go
@@ -188,6 +188,44 @@ func (s *retryActivitySuite) TestRetryActivity_should_be_scheduled_when_next_bac
 	s.assertTruncateFailureCalled()
 }
 
+// TestRetryActivity_should_be_scheduled_when_next_retry_delay_is_set asserts that the activity is retried after NextRetryDelay period specified in the application failure.
+func (s *retryActivitySuite) TestRetryActivity_should_be_scheduled_when_next_retry_delay_is_set() {
+	s.mutableState.timeSource = s.timeSource
+	taskGeneratorMock := NewMockTaskGenerator(s.controller)
+	nextAttempt := s.activity.Attempt + 1
+	expectedScheduledTime := s.timeSource.Now().Add(time.Minute).UTC()
+	taskGeneratorMock.EXPECT().GenerateActivityRetryTasks(s.activity.ScheduledEventId, expectedScheduledTime, nextAttempt)
+	s.mutableState.taskGenerator = taskGeneratorMock
+
+	s.failure.GetApplicationFailureInfo().NextRetryDelay = durationpb.New(time.Minute)
+	_, err := s.mutableState.RetryActivity(s.activity, s.failure)
+	s.NoError(err)
+	s.Equal(s.onActivityCreate.mutableStateApproximateSize-s.onActivityCreate.activitySize+s.activity.Size(), s.mutableState.approximateSize)
+	s.Equal(s.activity.Version, s.mutableState.currentVersion)
+	s.Equal(s.activity.Attempt, nextAttempt)
+
+	s.Equal(expectedScheduledTime, s.activity.ScheduledTime.AsTime(), "Activity scheduled time is incorrect")
+	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