Fix scheduler NPE for timer failover processor (#2831)

temporalio · May 10, 2022 · 19972ae · 19972ae
1 parent 668c851
commit 19972ae
Show file tree

Hide file tree

Showing 4 changed files with 11 additions and 3 deletions.
diff --git a/service/history/timerQueueActiveProcessor.go b/service/history/timerQueueActiveProcessor.go
@@ -145,6 +145,7 @@ func newTimerQueueActiveProcessor(
 func newTimerQueueFailoverProcessor(
 	shard shard.Context,
 	workflowCache workflow.Cache,
+	scheduler queues.Scheduler,
 	workflowDeleteManager workflow.DeleteManager,
 	namespaceIDs map[string]struct{},
 	standbyClusterName string,
@@ -202,7 +203,9 @@ func newTimerQueueFailoverProcessor(
 		matchingClient,
 	)
 
-	scheduler := newTimerTaskScheduler(shard, logger)
+	if scheduler == nil {
+		scheduler = newTimerTaskScheduler(shard, logger)
+	}
 
 	rescheduler := queues.NewRescheduler(
 		scheduler,
@@ -223,7 +226,7 @@ func newTimerQueueFailoverProcessor(
 				t,
 				timerTaskFilter,
 				taskExecutor,
-				nil,
+				scheduler,
 				rescheduler,
 				shard.GetTimeSource(),
 				logger,

diff --git a/service/history/timerQueueProcessor.go b/service/history/timerQueueProcessor.go
@@ -216,6 +216,7 @@ func (t *timerQueueProcessorImpl) FailoverNamespace(
 	updateShardAckLevel, failoverTimerProcessor := newTimerQueueFailoverProcessor(
 		t.shard,
 		t.workflowCache,
+		t.scheduler,
 		t.workflowDeleteManager,
 		namespaceIDs,
 		standbyClusterName,

diff --git a/service/history/transferQueueActiveProcessor.go b/service/history/transferQueueActiveProcessor.go
@@ -173,6 +173,7 @@ func newTransferQueueActiveProcessor(
 func newTransferQueueFailoverProcessor(
 	shard shard.Context,
 	workflowCache workflow.Cache,
+	scheduler queues.Scheduler,
 	archivalClient archiver.Client,
 	sdkClientFactory sdk.ClientFactory,
 	matchingClient matchingservice.MatchingServiceClient,
@@ -253,7 +254,9 @@ func newTransferQueueFailoverProcessor(
 		matchingClient,
 	)
 
-	scheduler := newTransferTaskScheduler(shard, logger)
+	if scheduler == nil {
+		scheduler = newTransferTaskScheduler(shard, logger)
+	}
 
 	rescheduler := queues.NewRescheduler(
 		scheduler,

diff --git a/service/history/transferQueueProcessor.go b/service/history/transferQueueProcessor.go
@@ -208,6 +208,7 @@ func (t *transferQueueProcessorImpl) FailoverNamespace(
 	updateShardAckLevel, failoverTaskProcessor := newTransferQueueFailoverProcessor(
 		t.shard,
 		t.workflowCache,
+		t.scheduler,
 		t.archivalClient,
 		t.sdkClientFactory,
 		t.matchingClient,