[history] add domain status check in taskfilter (uber#5140)

What changed?

add domain status check in taskAllocator verification

Why?

tasks in queues should drop for deprecated or deleted domains

service/history/queue/task_allocator.go

@@ -21,11 +21,13 @@
 package queue
 
 import (
+	"errors"
 	"sync"
 
 	"github.com/uber/cadence/common/cache"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
+	"github.com/uber/cadence/common/persistence"
 	"github.com/uber/cadence/common/types"
 	"github.com/uber/cadence/service/history/shard"
 	htask "github.com/uber/cadence/service/history/task"
@@ -189,3 +191,20 @@ func (t *taskAllocatorImpl) Lock() {
 func (t *taskAllocatorImpl) Unlock() {
 	t.locker.Unlock()
 }
+
+// isDomainNotRegistered checks either if domain does not exist or is in deprecated or deleted status
+func isDomainNotRegistered(shard shard.Context, domainID string) (bool, error) {
+	domainEntry, err := shard.GetDomainCache().GetDomainByID(domainID)
+	if err != nil {
+		if _, ok := err.(*types.EntityNotExistsError); ok {
+			return true, nil
+		}
+		// unexpected error in finding a domain
+		return false, err
+	}
+	info := domainEntry.GetInfo()
+	if info == nil {
+		return false, errors.New("domain info is nil in cache")
+	}
+	return info.Status == persistence.DomainStatusDeprecated || info.Status == persistence.DomainStatusDeleted, nil
+}

service/history/queue/timer_queue_processor.go

@@ -445,6 +445,11 @@ func newTimerQueueActiveProcessor(
 		if !ok {
 			return false, errUnexpectedQueueTask
 		}
+		if notRegistered, err := isDomainNotRegistered(shard, timer.DomainID); notRegistered && err == nil {
+			logger.Info("Domain is not in registered status, skip task in active timer queue.", tag.WorkflowDomainID(timer.DomainID), tag.Value(taskInfo))
+			return false, nil
+		}
+
 		return taskAllocator.VerifyActiveTask(timer.DomainID, timer)
 	}
 
@@ -502,6 +507,10 @@ func newTimerQueueStandbyProcessor(
 		if !ok {
 			return false, errUnexpectedQueueTask
 		}
+		if notRegistered, err := isDomainNotRegistered(shard, timer.DomainID); notRegistered && err == nil {
+			logger.Info("Domain is not in registered status, skip task in standby timer queue.", tag.WorkflowDomainID(timer.DomainID), tag.Value(taskInfo))
+			return false, nil
+		}
 		if timer.TaskType == persistence.TaskTypeWorkflowTimeout ||
 			timer.TaskType == persistence.TaskTypeDeleteHistoryEvent {
 			domainEntry, err := shard.GetDomainCache().GetDomainByID(timer.DomainID)
@@ -589,6 +598,10 @@ func newTimerQueueFailoverProcessor(
 		if !ok {
 			return false, errUnexpectedQueueTask
 		}
+		if notRegistered, err := isDomainNotRegistered(shardContext, timer.DomainID); notRegistered && err == nil {
+			logger.Info("Domain is not in registered status, skip task in failover timer queue.", tag.WorkflowDomainID(timer.DomainID), tag.Value(taskInfo))
+			return false, nil
+		}
 		return taskAllocator.VerifyFailoverActiveTask(domainIDs, timer.DomainID, timer)
 	}
 

service/history/queue/transfer_queue_processor.go

@@ -458,6 +458,10 @@ func newTransferQueueActiveProcessor(
 		if !ok {
 			return false, errUnexpectedQueueTask
 		}
+		if notRegistered, err := isDomainNotRegistered(shard, task.DomainID); notRegistered && err == nil {
+			logger.Info("Domain is not in registered status, skip task in active transfer queue.", tag.WorkflowDomainID(task.DomainID), tag.Value(taskInfo))
+			return false, nil
+		}
 		return taskAllocator.VerifyActiveTask(task.DomainID, task)
 	}
 
@@ -514,6 +518,10 @@ func newTransferQueueStandbyProcessor(
 		if !ok {
 			return false, errUnexpectedQueueTask
 		}
+		if notRegistered, err := isDomainNotRegistered(shard, task.DomainID); notRegistered && err == nil {
+			logger.Info("Domain is not in registered status, skip task in standby transfer queue.", tag.WorkflowDomainID(task.DomainID), tag.Value(taskInfo))
+			return false, nil
+		}
 		if task.TaskType == persistence.TransferTaskTypeCloseExecution ||
 			task.TaskType == persistence.TransferTaskTypeRecordWorkflowClosed {
 			domainEntry, err := shard.GetDomainCache().GetDomainByID(task.DomainID)
@@ -596,6 +604,10 @@ func newTransferQueueFailoverProcessor(
 		if !ok {
 			return false, errUnexpectedQueueTask
 		}
+		if notRegistered, err := isDomainNotRegistered(shardContext, task.DomainID); notRegistered && err == nil {
+			logger.Info("Domain is not in registered status, skip task in failover transfer queue.", tag.WorkflowDomainID(task.DomainID), tag.Value(taskInfo))
+			return false, nil
+		}
 		return taskAllocator.VerifyFailoverActiveTask(domainIDs, task.DomainID, task)
 	}