Adjust scan intervals for transfer queue processor (#356)

Author: samarabbas

service/history/service.go

@@ -53,13 +53,15 @@ type Config struct {
 	TimerProcessorUpdateFailureRetryCount int
 	TimerProcessorGetFailureRetryCount    int
 	TimerProcessorUpdateAckInterval       time.Duration
+	TimerProcessorForceUpdateInterval     time.Duration
 
 	// TransferQueueProcessor settings
-	TransferTaskBatchSize              int
-	TransferProcessorMaxPollRPS        int
-	TransferProcessorMaxPollInterval   time.Duration
-	TransferProcessorUpdateAckInterval time.Duration
-	TransferTaskWorkerCount            int
+	TransferTaskBatchSize                int
+	TransferProcessorMaxPollRPS          int
+	TransferProcessorMaxPollInterval     time.Duration
+	TransferProcessorUpdateAckInterval   time.Duration
+	TransferProcessorForceUpdateInterval time.Duration
+	TransferTaskWorkerCount              int
 }
 
 // NewConfig returns new service config with default values
@@ -79,10 +81,12 @@ func NewConfig(numberOfShards int) *Config {
 		TimerProcessorUpdateFailureRetryCount:       5,
 		TimerProcessorGetFailureRetryCount:          5,
 		TimerProcessorUpdateAckInterval:             10 * time.Second,
+		TimerProcessorForceUpdateInterval:           10 * time.Minute,
 		TransferTaskBatchSize:                       10,
 		TransferProcessorMaxPollRPS:                 100,
-		TransferProcessorMaxPollInterval:            10 * time.Second,
+		TransferProcessorMaxPollInterval:            60 * time.Second,
 		TransferProcessorUpdateAckInterval:          10 * time.Second,
+		TransferProcessorForceUpdateInterval:        10 * time.Minute,
 		TransferTaskWorkerCount:                     10,
 	}
 }

service/history/timerQueueProcessor.go

@@ -81,6 +81,8 @@ type (
 		readLevel        SequenceID
 		ackLevel         time.Time
 		metricsClient    metrics.Client
+		lastUpdated      time.Time
+		config           *Config
 	}
 )
 
@@ -890,6 +892,7 @@ func (t timerTaskIDs) Less(i, j int) bool {
 func newTimerAckMgr(processor *timerQueueProcessorImpl, shard ShardContext, executionMgr persistence.ExecutionManager,
 	logger bark.Logger) *timerAckMgr {
 	ackLevel := shard.GetTimerAckLevel()
+	config := shard.GetConfig()
 	return &timerAckMgr{
 		processor:        processor,
 		shard:            shard,
@@ -899,6 +902,8 @@ func newTimerAckMgr(processor *timerQueueProcessorImpl, shard ShardContext, exec
 		ackLevel:         ackLevel,
 		metricsClient:    processor.metricsClient,
 		logger:           logger,
+		lastUpdated:      time.Now(),
+		config:           config,
 	}
 }
 
@@ -957,6 +962,7 @@ func (t *timerAckMgr) completeTimerTask(taskID SequenceID) {
 
 func (t *timerAckMgr) updateAckLevel() {
 	t.metricsClient.IncCounter(metrics.TimerQueueProcessorScope, metrics.AckLevelUpdateCounter)
+	initialAckLevel := t.ackLevel
 	updatedAckLevel := t.ackLevel
 	t.Lock()
 
@@ -982,6 +988,11 @@ MoveAckLevelLoop:
 	}
 	t.Unlock()
 
+	// Do not update Acklevel if nothing changed upto force update interval
+	if initialAckLevel == updatedAckLevel && time.Since(t.lastUpdated) > t.config.TimerProcessorForceUpdateInterval {
+		return
+	}
+
 	t.logger.Debugf("Updating timer ack level: %v", updatedAckLevel)
 
 	// Always update ackLevel to detect if the shared is stolen

service/history/transferQueueProcessor.go

@@ -73,6 +73,8 @@ type (
 		executionMgr  persistence.ExecutionManager
 		logger        bark.Logger
 		metricsClient metrics.Client
+		lastUpdated   time.Time
+		config        *Config
 
 		sync.RWMutex
 		outstandingTasks map[int64]bool
@@ -112,6 +114,7 @@ func newTransferQueueProcessor(shard ShardContext, visibilityMgr persistence.Vis
 func newAckManager(processor *transferQueueProcessorImpl, shard ShardContext, executionMgr persistence.ExecutionManager,
 	logger bark.Logger, metricsClient metrics.Client) *ackManager {
 	ackLevel := shard.GetTransferAckLevel()
+	config := shard.GetConfig()
 	return &ackManager{
 		processor:        processor,
 		shard:            shard,
@@ -121,6 +124,8 @@ func newAckManager(processor *transferQueueProcessorImpl, shard ShardContext, ex
 		ackLevel:         ackLevel,
 		logger:           logger,
 		metricsClient:    metricsClient,
+		lastUpdated:      time.Now(),
+		config:           config,
 	}
 }
 
@@ -912,6 +917,7 @@ func (a *ackManager) completeTask(taskID int64) {
 
 func (a *ackManager) updateAckLevel() {
 	a.metricsClient.IncCounter(metrics.TransferQueueProcessorScope, metrics.AckLevelUpdateCounter)
+	initialAckLevel := a.ackLevel
 	updatedAckLevel := a.ackLevel
 	a.Lock()
 MoveAckLevelLoop:
@@ -935,6 +941,11 @@ MoveAckLevelLoop:
 	}
 	a.Unlock()
 
+	// Do not update Acklevel if nothing changed upto force update interval
+	if initialAckLevel == updatedAckLevel && time.Since(a.lastUpdated) > a.config.TransferProcessorForceUpdateInterval {
+		return
+	}
+
 	// Always update ackLevel to detect if the shared is stolen
 	if err := a.shard.UpdateTransferAckLevel(updatedAckLevel); err != nil {
 		a.metricsClient.IncCounter(metrics.TransferQueueProcessorScope, metrics.AckLevelUpdateFailedCounter)