diff --git a/service/history/queues/executable.go b/service/history/queues/executable.go
index c8be0a88431..54457d6c73a 100644
--- a/service/history/queues/executable.go
+++ b/service/history/queues/executable.go
@@ -237,12 +237,6 @@ func (e *executableImpl) HandleErr(err error) (retErr error) {
 				e.taggedMetricsHandler.Histogram(TaskAttempt, metrics.Dimensionless).Record(int64(e.attempt))
 				e.logger.Error("Critical error processing task, retrying.", tag.Error(err), tag.OperationCritical)
 			}
-
-			if common.IsResourceExhausted(retErr) {
-				e.resourceExhaustedCount++
-			} else {
-				e.resourceExhaustedCount = 0
-			}
 		}
 	}()
 
@@ -250,6 +244,13 @@ func (e *executableImpl) HandleErr(err error) (retErr error) {
 		return nil
 	}
 
+	if common.IsResourceExhausted(err) {
+		e.resourceExhaustedCount++
+		e.taggedMetricsHandler.Counter(TaskThrottledCounter).Record(1)
+		return err
+	}
+	e.resourceExhaustedCount = 0
+
 	if _, isNotFound := err.(*serviceerror.NotFound); isNotFound {
 		return nil
 	}
diff --git a/service/history/queues/metrics.go b/service/history/queues/metrics.go
index fa60ef49fef..ba8df92332b 100644
--- a/service/history/queues/metrics.go
+++ b/service/history/queues/metrics.go
@@ -43,6 +43,7 @@ const (
 	TaskStandbyRetryCounter         = "task_errors_standby_retry_counter"
 	TaskWorkflowBusyCounter         = "task_errors_workflow_busy"
 	TaskNotActiveCounter            = "task_errors_not_active_counter"
+	TaskThrottledCounter            = "task_errors_throttled"
 	TaskLoadLatency                 = "task_latency_load"                     // latency from task generation to task loading (persistence scheduleToStart)
 	TaskScheduleLatency             = "task_latency_schedule"                 // latency from task submission to in-memory queue to processing (in-memory scheduleToStart)
 	TaskProcessingLatency           = "task_latency_processing"               // latency for processing task one time
@@ -53,7 +54,6 @@ const (
 	TaskNoUserQueueLatency          = "task_latency_queue_nouserlatency"      // same as TaskQueueLatency, but excludes workflow lock latency
 	TaskUserLatency                 = "task_latency_userlatency"              // workflow lock latency across multiple attempts
 	TaskReschedulerPendingTasks     = "task_rescheduler_pending_tasks"
-	TaskThrottledCounter            = "task_throttled_counter"
 	TasksDependencyTaskNotCompleted = "task_dependency_task_not_completed"
 
 	TaskBatchCompleteCounter    = "task_batch_complete_counter"