Do not submit corrupted history tasks to scheduler (#6999)

Author: Shaddoll

common/metrics/defs.go

@@ -2407,6 +2407,7 @@ const (
 	ProcessingQueueSelectedDomainSplitCounter
 	ProcessingQueueRandomSplitCounter
 	ProcessingQueueThrottledCounter
+	CorruptedHistoryTaskCounter
 
 	QueueValidatorLostTaskCounter
 	QueueValidatorDropTaskCounter
@@ -3157,6 +3158,7 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		ProcessingQueueSelectedDomainSplitCounter:                    {metricName: "processing_queue_selected_domain_split_counter", metricType: Counter},
 		ProcessingQueueRandomSplitCounter:                            {metricName: "processing_queue_random_split_counter", metricType: Counter},
 		ProcessingQueueThrottledCounter:                              {metricName: "processing_queue_throttled_counter", metricType: Counter},
+		CorruptedHistoryTaskCounter:                                  {metricName: "corrupted_history_task_counter", metricType: Counter},
 		QueueValidatorLostTaskCounter:                                {metricName: "queue_validator_lost_task_counter", metricType: Counter},
 		QueueValidatorDropTaskCounter:                                {metricName: "queue_validator_drop_task_counter", metricType: Counter},
 		QueueValidatorInvalidLoadCounter:                             {metricName: "queue_validator_invalid_load_counter", metricType: Counter},

service/history/queue/processing_queue.go

@@ -28,7 +28,6 @@ import (
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/common/metrics"
-	"github.com/uber/cadence/common/persistence"
 	t "github.com/uber/cadence/common/task"
 	"github.com/uber/cadence/service/history/task"
 )
@@ -192,11 +191,6 @@ func (q *processingQueueImpl) AddTasks(tasks map[task.Key]task.Task, newReadLeve
 			continue
 		}
 
-		if persistence.IsTaskCorrupted(task) {
-			q.logger.Error("Processing queue encountered a corrupted task", tag.Dynamic("task", task))
-			continue
-		}
-
 		if !taskBelongsToProcessQueue(q.state, key, task) {
 			errMsg := "Processing queue encountered a task doesn't belong to its scope"
 			q.logger.Error(errMsg, tag.Error(

service/history/queue/timer_queue_processor_base.go

@@ -305,6 +305,12 @@ func (t *timerQueueProcessorBase) processQueueCollections(levels map[int]struct{
 				continue
 			}
 
+			if persistence.IsTaskCorrupted(taskInfo) {
+				t.logger.Error("Processing queue encountered a corrupted task", tag.Dynamic("task", taskInfo))
+				t.metricsScope.IncCounter(metrics.CorruptedHistoryTaskCounter)
+				continue
+			}
+
 			task := t.taskInitializer(taskInfo)
 			tasks[newTimerTaskKey(taskInfo.GetVisibilityTimestamp(), taskInfo.GetTaskID())] = task
 			submitted, err := t.submitTask(task)

service/history/queue/transfer_queue_processor_base.go

@@ -428,6 +428,12 @@ func (t *transferQueueProcessorBase) processQueueCollections() {
 				continue
 			}
 
+			if persistence.IsTaskCorrupted(taskInfo) {
+				t.logger.Error("Processing queue encountered a corrupted task", tag.Dynamic("task", taskInfo))
+				t.metricsScope.IncCounter(metrics.CorruptedHistoryTaskCounter)
+				continue
+			}
+
 			task := t.taskInitializer(taskInfo)
 			tasks[newTransferTaskKey(taskInfo.GetTaskID())] = task
 			submitted, err := t.submitTask(task)

service/history/queue/transfer_queue_processor_base_test.go

@@ -354,7 +354,9 @@ func (s *transferQueueProcessorBaseSuite) TestProcessQueueCollections_WithNextPa
 	taskInfos := []persistence.Task{
 		&persistence.DecisionTask{
 			WorkflowIdentifier: persistence.WorkflowIdentifier{
-				DomainID: "testDomain1",
+				DomainID:   "testDomain1",
+				WorkflowID: "testWorkflowID",
+				RunID:      "testRunID",
 			},
 			TaskData: persistence.TaskData{
 				TaskID: 500,

service/history/queuev2/virtual_queue.go

@@ -243,6 +243,8 @@ func (q *virtualQueueImpl) loadAndSubmitTasks() {
 	for _, task := range tasks {
 		if persistence.IsTaskCorrupted(task) {
 			q.logger.Error("Virtual queue encountered a corrupted task", tag.Dynamic("task", task))
+			q.metricsScope.IncCounter(metrics.CorruptedHistoryTaskCounter)
+			task.Ack()
 			continue
 		}