Refactor history queue factory

Author: Shaddoll

service/history/engine/engineimpl/describe_queues.go

@@ -25,6 +25,7 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/uber/cadence/common/persistence"
 	"github.com/uber/cadence/common/types"
 	"github.com/uber/cadence/service/history/queue"
 )
@@ -33,21 +34,25 @@ func (e *historyEngineImpl) DescribeTransferQueue(
 	ctx context.Context,
 	clusterName string,
 ) (*types.DescribeQueueResponse, error) {
-	return e.describeQueue(ctx, e.txProcessor, clusterName)
+	return e.describeQueue(ctx, persistence.HistoryTaskCategoryTransfer, clusterName)
 }
 
 func (e *historyEngineImpl) DescribeTimerQueue(
 	ctx context.Context,
 	clusterName string,
 ) (*types.DescribeQueueResponse, error) {
-	return e.describeQueue(ctx, e.timerProcessor, clusterName)
+	return e.describeQueue(ctx, persistence.HistoryTaskCategoryTimer, clusterName)
 }
 
 func (e *historyEngineImpl) describeQueue(
 	ctx context.Context,
-	queueProcessor queue.Processor,
+	category persistence.HistoryTaskCategory,
 	clusterName string,
 ) (*types.DescribeQueueResponse, error) {
+	queueProcessor, ok := e.queueProcessors[category]
+	if !ok {
+		return nil, fmt.Errorf("queue processor not found for category %v", category)
+	}
 	resp, err := queueProcessor.HandleAction(ctx, clusterName, queue.NewGetStateAction())
 	if err != nil {
 		return nil, err

service/history/engine/engineimpl/history_engine.go

@@ -27,8 +27,6 @@ import (
 	"fmt"
 	"time"
 
-	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
-
 	"github.com/uber/cadence/client/matching"
 	"github.com/uber/cadence/client/wrappers/retryable"
 	"github.com/uber/cadence/common"
@@ -43,10 +41,7 @@ import (
 	"github.com/uber/cadence/common/metrics"
 	cndc "github.com/uber/cadence/common/ndc"
 	"github.com/uber/cadence/common/persistence"
-	"github.com/uber/cadence/common/quotas"
-	"github.com/uber/cadence/common/quotas/permember"
 	"github.com/uber/cadence/common/reconciliation/invariant"
-	"github.com/uber/cadence/common/service"
 	"github.com/uber/cadence/common/types"
 	"github.com/uber/cadence/common/types/mapper/proto"
 	hcommon "github.com/uber/cadence/service/history/common"
@@ -61,10 +56,8 @@ import (
 	"github.com/uber/cadence/service/history/replication"
 	"github.com/uber/cadence/service/history/reset"
 	"github.com/uber/cadence/service/history/shard"
-	"github.com/uber/cadence/service/history/task"
 	"github.com/uber/cadence/service/history/workflow"
-	"github.com/uber/cadence/service/history/workflowcache"
-	warchiver "github.com/uber/cadence/service/worker/archiver"
+	"github.com/uber/cadence/service/worker/archiver"
 )
 
 const (
@@ -92,8 +85,7 @@ type historyEngineImpl struct {
 	historyV2Mgr              persistence.HistoryManager
 	executionManager          persistence.ExecutionManager
 	visibilityMgr             persistence.VisibilityManager
-	txProcessor               queue.Processor
-	timerProcessor            queue.Processor
+	queueProcessors           map[persistence.HistoryTaskCategory]queue.Processor
 	nDCReplicator             ndc.HistoryReplicator
 	nDCActivityReplicator     ndc.ActivityReplicator
 	historyEventNotifier      events.Notifier
@@ -104,22 +96,19 @@ type historyEngineImpl struct {
 	throttledLogger           log.Logger
 	activeClusterManager      activecluster.Manager
 	config                    *config.Config
-	archivalClient            warchiver.Client
+	archivalClient            archiver.Client
 	workflowResetter          reset.WorkflowResetter
-	queueTaskProcessor        task.Processor
 	replicationTaskProcessors []replication.TaskProcessor
 	replicationAckManager     replication.TaskAckManager
 	replicationTaskStore      *replication.TaskStore
 	replicationHydrator       replication.TaskHydrator
 	replicationMetricsEmitter *replication.MetricsEmitterImpl
-	publicClient              workflowserviceclient.Interface
 	eventsReapplier           ndc.EventsReapplier
 	matchingClient            matching.Client
 	rawMatchingClient         matching.Client
 	clientChecker             client.VersionChecker
 	replicationDLQHandler     replication.DLQHandler
 	failoverMarkerNotifier    failover.MarkerNotifier
-	wfIDCache                 workflowcache.WFCache
 
 	updateWithActionFn func(
 		context.Context,
@@ -149,15 +138,12 @@ func NewEngineWithShardContext(
 	shard shard.Context,
 	visibilityMgr persistence.VisibilityManager,
 	matching matching.Client,
-	publicClient workflowserviceclient.Interface,
 	historyEventNotifier events.Notifier,
 	config *config.Config,
 	replicationTaskFetchers replication.TaskFetchers,
 	rawMatchingClient matching.Client,
-	queueTaskProcessor task.Processor,
 	failoverCoordinator failover.Coordinator,
-	wfIDCache workflowcache.WFCache,
-	queueProcessorFactory queue.ProcessorFactory,
+	queueFactories []queue.Factory,
 ) engine.Engine {
 	currentClusterName := shard.GetService().GetClusterMetadata().GetCurrentClusterName()
 
@@ -194,40 +180,13 @@ func NewEngineWithShardContext(
 		metricsClient:        shard.GetMetricsClient(),
 		historyEventNotifier: historyEventNotifier,
 		config:               config,
-		archivalClient: warchiver.NewClient(
-			shard.GetMetricsClient(),
-			logger,
-			publicClient,
-			shard.GetConfig().NumArchiveSystemWorkflows,
-			quotas.NewDynamicRateLimiter(config.ArchiveRequestRPS.AsFloat64()),
-			quotas.NewDynamicRateLimiter(func() float64 {
-				return permember.PerMember(
-					service.History,
-					float64(config.ArchiveInlineHistoryGlobalRPS()),
-					float64(config.ArchiveInlineHistoryRPS()),
-					shard.GetService().GetMembershipResolver(),
-				)
-			}),
-			quotas.NewDynamicRateLimiter(func() float64 {
-				return permember.PerMember(
-					service.History,
-					float64(config.ArchiveInlineVisibilityGlobalRPS()),
-					float64(config.ArchiveInlineVisibilityRPS()),
-					shard.GetService().GetMembershipResolver(),
-				)
-			}),
-			shard.GetService().GetArchiverProvider(),
-			config.AllowArchivingIncompleteHistory,
-		),
 		workflowResetter: reset.NewWorkflowResetter(
 			shard,
 			executionCache,
 			logger,
 		),
-		publicClient:           publicClient,
 		matchingClient:         matching,
 		rawMatchingClient:      rawMatchingClient,
-		queueTaskProcessor:     queueTaskProcessor,
 		clientChecker:          client.NewVersionChecker(),
 		failoverMarkerNotifier: failoverMarkerNotifier,
 		replicationHydrator:    replicationHydrator,
@@ -246,8 +205,8 @@ func NewEngineWithShardContext(
 		replicationTaskStore: replicationTaskStore,
 		replicationMetricsEmitter: replication.NewMetricsEmitter(
 			shard.GetShardID(), shard, replicationReader, shard.GetMetricsClient()),
-		wfIDCache:          wfIDCache,
 		updateWithActionFn: workflow.UpdateWithAction,
+		queueProcessors:    make(map[persistence.HistoryTaskCategory]queue.Processor),
 	}
 	historyEngImpl.decisionHandler = decision.NewHandler(
 		shard,
@@ -261,25 +220,13 @@ func NewEngineWithShardContext(
 	)
 	openExecutionCheck := invariant.NewConcreteExecutionExists(pRetry, shard.GetDomainCache())
 
-	historyEngImpl.txProcessor = queueProcessorFactory.NewTransferQueueProcessor(
-		shard,
-		historyEngImpl,
-		queueTaskProcessor,
-		executionCache,
-		historyEngImpl.workflowResetter,
-		historyEngImpl.archivalClient,
-		openExecutionCheck,
-		historyEngImpl.wfIDCache,
-	)
-
-	historyEngImpl.timerProcessor = queueProcessorFactory.NewTimerQueueProcessor(
-		shard,
-		historyEngImpl,
-		queueTaskProcessor,
-		executionCache,
-		historyEngImpl.archivalClient,
-		openExecutionCheck,
-	)
+	for _, factory := range queueFactories {
+		historyEngImpl.queueProcessors[factory.Category()] = factory.CreateQueue(
+			shard,
+			executionCache,
+			openExecutionCheck,
+		)
+	}
 
 	historyEngImpl.eventsReapplier = ndc.NewEventsReapplier(shard.GetMetricsClient(), logger)
 
@@ -360,8 +307,9 @@ func (e *historyEngineImpl) Start() {
 	e.logger.Info("History engine state changed", tag.LifeCycleStarting)
 	defer e.logger.Info("History engine state changed", tag.LifeCycleStarted)
 
-	e.txProcessor.Start()
-	e.timerProcessor.Start()
+	for _, processor := range e.queueProcessors {
+		processor.Start()
+	}
 	e.replicationDLQHandler.Start()
 	e.replicationMetricsEmitter.Start()
 
@@ -386,8 +334,9 @@ func (e *historyEngineImpl) Stop() {
 	e.logger.Info("History engine state changed", tag.LifeCycleStopping)
 	defer e.logger.Info("History engine state changed", tag.LifeCycleStopped)
 
-	e.txProcessor.Stop()
-	e.timerProcessor.Stop()
+	for _, processor := range e.queueProcessors {
+		processor.Stop()
+	}
 	e.replicationDLQHandler.Stop()
 	e.replicationMetricsEmitter.Stop()
 
@@ -420,8 +369,9 @@ func (e *historyEngineImpl) SyncShardStatus(ctx context.Context, request *types.
 	// 2. notify the timer gate in the timer queue standby processor
 	// 3. notify the transfer (essentially a no op, just put it here so it looks symmetric)
 	e.shard.SetCurrentTime(clusterName, now)
-	e.txProcessor.NotifyNewTask(clusterName, &hcommon.NotifyTaskInfo{Tasks: []persistence.Task{}})
-	e.timerProcessor.NotifyNewTask(clusterName, &hcommon.NotifyTaskInfo{Tasks: []persistence.Task{}})
+	for _, processor := range e.queueProcessors {
+		processor.NotifyNewTask(clusterName, &hcommon.NotifyTaskInfo{Tasks: []persistence.Task{}})
+	}
 	return nil
 }
 

service/history/engine/engineimpl/history_engine2_test.go

@@ -150,8 +150,10 @@ func (s *engine2Suite) SetupTest() {
 		config:               s.config,
 		timeSource:           s.mockShard.GetTimeSource(),
 		historyEventNotifier: events.NewNotifier(clock.NewRealTimeSource(), metrics.NewClient(tally.NoopScope, metrics.History), func(string) int { return 0 }),
-		txProcessor:          s.mockTxProcessor,
-		timerProcessor:       s.mockTimerProcessor,
+		queueProcessors: map[p.HistoryTaskCategory]queue.Processor{
+			p.HistoryTaskCategoryTransfer: s.mockTxProcessor,
+			p.HistoryTaskCategoryTimer:    s.mockTimerProcessor,
+		},
 	}
 	s.mockShard.SetEngine(h)
 	h.decisionHandler = decision.NewHandler(s.mockShard, h.executionCache, h.tokenSerializer)

service/history/engine/engineimpl/history_engine3_eventsv2_test.go

@@ -133,8 +133,10 @@ func (s *engine3Suite) SetupTest() {
 		config:               s.config,
 		timeSource:           s.mockShard.GetTimeSource(),
 		historyEventNotifier: events.NewNotifier(clock.NewRealTimeSource(), metrics.NewClient(tally.NoopScope, metrics.History), func(string) int { return 0 }),
-		txProcessor:          s.mockTxProcessor,
-		timerProcessor:       s.mockTimerProcessor,
+		queueProcessors: map[p.HistoryTaskCategory]queue.Processor{
+			p.HistoryTaskCategoryTransfer: s.mockTxProcessor,
+			p.HistoryTaskCategoryTimer:    s.mockTimerProcessor,
+		},
 	}
 	s.mockShard.SetEngine(h)
 	h.decisionHandler = decision.NewHandler(s.mockShard, h.executionCache, h.tokenSerializer)

service/history/engine/engineimpl/history_engine_test.go

@@ -173,11 +173,13 @@ func (s *engineSuite) SetupTest() {
 		tokenSerializer:      common.NewJSONTaskTokenSerializer(),
 		historyEventNotifier: historyEventNotifier,
 		config:               config.NewForTest(),
-		txProcessor:          s.mockTxProcessor,
-		timerProcessor:       s.mockTimerProcessor,
-		clientChecker:        cc.NewVersionChecker(),
-		eventsReapplier:      s.mockEventsReapplier,
-		workflowResetter:     s.mockWorkflowResetter,
+		queueProcessors: map[persistence.HistoryTaskCategory]queue.Processor{
+			persistence.HistoryTaskCategoryTransfer: s.mockTxProcessor,
+			persistence.HistoryTaskCategoryTimer:    s.mockTimerProcessor,
+		},
+		clientChecker:    cc.NewVersionChecker(),
+		eventsReapplier:  s.mockEventsReapplier,
+		workflowResetter: s.mockWorkflowResetter,
 	}
 	s.mockShard.SetEngine(h)
 	h.decisionHandler = decision.NewHandler(s.mockShard, h.executionCache, h.tokenSerializer)

service/history/engine/engineimpl/notify_tasks.go

@@ -46,7 +46,12 @@ func (e *historyEngineImpl) NotifyNewTransferTasks(info *hcommon.NotifyTaskInfo)
 	task := info.Tasks[0]
 	clusterName, err := e.shard.GetActiveClusterManager().ClusterNameForFailoverVersion(task.GetVersion(), info.ExecutionInfo.DomainID)
 	if err == nil {
-		e.txProcessor.NotifyNewTask(clusterName, info)
+		transferProcessor, ok := e.queueProcessors[persistence.HistoryTaskCategoryTransfer]
+		if !ok {
+			e.logger.Error("transfer processor not found", tag.Error(err))
+			return
+		}
+		transferProcessor.NotifyNewTask(clusterName, info)
 	}
 }
 
@@ -58,7 +63,12 @@ func (e *historyEngineImpl) NotifyNewTimerTasks(info *hcommon.NotifyTaskInfo) {
 	task := info.Tasks[0]
 	clusterName, err := e.shard.GetActiveClusterManager().ClusterNameForFailoverVersion(task.GetVersion(), info.ExecutionInfo.DomainID)
 	if err == nil {
-		e.timerProcessor.NotifyNewTask(clusterName, info)
+		timerProcessor, ok := e.queueProcessors[persistence.HistoryTaskCategoryTimer]
+		if !ok {
+			e.logger.Error("timer processor not found", tag.Error(err))
+			return
+		}
+		timerProcessor.NotifyNewTask(clusterName, info)
 	}
 }
 

service/history/engine/engineimpl/register_domain_failover_callback.go

@@ -127,8 +127,9 @@ func (e *historyEngineImpl) domainChangeCB(nextDomains []*cache.DomainCacheEntry
 		e.logger.Info("Domain Failover Start.", tag.WorkflowDomainIDs(failoverDomainIDs))
 
 		// Failover queues are not created for active-active domains. Will revisit after new queue framework implementation.
-		e.txProcessor.FailoverDomain(failoverDomainIDs)
-		e.timerProcessor.FailoverDomain(failoverDomainIDs)
+		for _, processor := range e.queueProcessors {
+			processor.FailoverDomain(failoverDomainIDs)
+		}
 
 		e.notifyQueues()
 	}
@@ -159,8 +160,18 @@ func (e *historyEngineImpl) notifyQueues() {
 	// its length > 0 and has correct timestamp, to trigger a db scan
 	fakeDecisionTask := []persistence.Task{&persistence.DecisionTask{}}
 	fakeDecisionTimeoutTask := []persistence.Task{&persistence.DecisionTimeoutTask{TaskData: persistence.TaskData{VisibilityTimestamp: now}}}
-	e.txProcessor.NotifyNewTask(e.currentClusterName, &hcommon.NotifyTaskInfo{Tasks: fakeDecisionTask})
-	e.timerProcessor.NotifyNewTask(e.currentClusterName, &hcommon.NotifyTaskInfo{Tasks: fakeDecisionTimeoutTask})
+	transferProcessor, ok := e.queueProcessors[persistence.HistoryTaskCategoryTransfer]
+	if !ok {
+		e.logger.Error("transfer processor not found")
+		return
+	}
+	transferProcessor.NotifyNewTask(e.currentClusterName, &hcommon.NotifyTaskInfo{Tasks: fakeDecisionTask})
+	timerProcessor, ok := e.queueProcessors[persistence.HistoryTaskCategoryTimer]
+	if !ok {
+		e.logger.Error("timer processor not found")
+		return
+	}
+	timerProcessor.NotifyNewTask(e.currentClusterName, &hcommon.NotifyTaskInfo{Tasks: fakeDecisionTimeoutTask})
 }
 
 func (e *historyEngineImpl) generateGracefulFailoverTasksForDomainUpdateCallback(shardNotificationVersion int64, nextDomains []*cache.DomainCacheEntry) []*persistence.FailoverMarkerTask {
@@ -204,14 +215,16 @@ func (e *historyEngineImpl) generateGracefulFailoverTasksForDomainUpdateCallback
 
 func (e *historyEngineImpl) lockTaskProcessingForDomainUpdate() {
 	e.logger.Debug("Locking processing for domain update")
-	e.txProcessor.LockTaskProcessing()
-	e.timerProcessor.LockTaskProcessing()
+	for _, processor := range e.queueProcessors {
+		processor.LockTaskProcessing()
+	}
 }
 
 func (e *historyEngineImpl) unlockProcessingForDomainUpdate() {
 	e.logger.Debug("Unlocking processing for failover")
-	e.txProcessor.UnlockTaskProcessing()
-	e.timerProcessor.UnlockTaskProcessing()
+	for _, processor := range e.queueProcessors {
+		processor.UnlockTaskProcessing()
+	}
 }
 
 func (e *historyEngineImpl) failoverPredicate(shardNotificationVersion int64, nextDomain *cache.DomainCacheEntry, action func()) {