01875e2f-959c-4c4d-87af-1d7805759bcc

Committed 08 Apr 2023 12:26AM UTC coverage: 57.178% (+0.1%) from 57.072%

Build # 01875e2f-959c-4c4d-87af-1d7805759bcc

Build Type

buildkite

Committed by Steven L

Commit Message

bad cleanup -> good cleanup

Pull Request Pull Request #5197: Demonstrate a way to get rid of the cadence-idl repo

Run Details

85396 of 149351 relevant lines covered (57.18%)

2283.28 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

78.81

/service/history/queue/timer_queue_processor_base.go

// Copyright (c) 2017-2020 Uber Technologies Inc.

// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:

// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.

// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.

package queue

import (
        "context"
        "fmt"
        "math"
        "math/rand"
        "sync"
        "sync/atomic"
        "time"

        "github.com/uber/cadence/common"
        "github.com/uber/cadence/common/backoff"
        "github.com/uber/cadence/common/dynamicconfig"
        "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"
        "github.com/uber/cadence/service/history/config"
        "github.com/uber/cadence/service/history/shard"
        "github.com/uber/cadence/service/history/task"
)

var (
        maximumTimerTaskKey = newTimerTaskKey(
                time.Unix(0, math.MaxInt64),
                0,
        )
)

type (
        timerTaskKey struct {
                visibilityTimestamp time.Time
                taskID              int64
        }

        timeTaskReadProgress struct {
                currentQueue  ProcessingQueue
                readLevel     task.Key
                maxReadLevel  task.Key
                nextPageToken []byte
        }

        timerQueueProcessorBase struct {
                *processorBase

                taskInitializer task.Initializer

                clusterName string

                pollTimeLock sync.Mutex
                backoffTimer map[int]*time.Timer
                nextPollTime map[int]time.Time
                timerGate    TimerGate

                // timer notification
                newTimerCh  chan struct{}
                newTimeLock sync.Mutex
                newTime     time.Time

                processingQueueReadProgress map[int]timeTaskReadProgress
        }
)

func newTimerQueueProcessorBase(
        clusterName string,
        shard shard.Context,
        processingQueueStates []ProcessingQueueState,
        taskProcessor task.Processor,
        timerGate TimerGate,
        options *queueProcessorOptions,
        updateMaxReadLevel updateMaxReadLevelFn,
        updateClusterAckLevel updateClusterAckLevelFn,
        updateProcessingQueueStates updateProcessingQueueStatesFn,
        queueShutdown queueShutdownFn,
        taskFilter task.Filter,
        taskExecutor task.Executor,
        logger log.Logger,
        metricsClient metrics.Client,
) *timerQueueProcessorBase {
        processorBase := newProcessorBase(
                shard,
                processingQueueStates,
                taskProcessor,
                options,
                updateMaxReadLevel,
                updateClusterAckLevel,
                updateProcessingQueueStates,
                queueShutdown,
                logger.WithTags(tag.ComponentTimerQueue),
                metricsClient,
        )

        queueType := task.QueueTypeActiveTimer
        if options.MetricScope == metrics.TimerStandbyQueueProcessorScope {
                queueType = task.QueueTypeStandbyTimer
        }

        return &timerQueueProcessorBase{
                processorBase: processorBase,

                taskInitializer: func(taskInfo task.Info) task.Task {
                        return task.NewTimerTask(
                                shard,
                                taskInfo,
                                queueType,
                                task.InitializeLoggerForTask(shard.GetShardID(), taskInfo, logger),
                                taskFilter,
                                taskExecutor,
                                taskProcessor,
                                processorBase.redispatcher.AddTask,
                                shard.GetConfig().TaskCriticalRetryCount,
                        )
                },

                clusterName: clusterName,

                backoffTimer: make(map[int]*time.Timer),
                nextPollTime: make(map[int]time.Time),
                timerGate:    timerGate,

                newTimerCh: make(chan struct{}, 1),

                processingQueueReadProgress: make(map[int]timeTaskReadProgress),
        }
}

func (t *timerQueueProcessorBase) Start() {
        if !atomic.CompareAndSwapInt32(&t.status, common.DaemonStatusInitialized, common.DaemonStatusStarted) {
                return
        }

        t.logger.Info("Timer queue processor state changed", tag.LifeCycleStarting)
        defer t.logger.Info("Timer queue processor state changed", tag.LifeCycleStarted)

        t.redispatcher.Start()

        newPollTime := time.Time{}
        if startJitter := t.options.MaxStartJitterInterval(); startJitter > 0 {
                now := t.shard.GetTimeSource().Now()
                newPollTime = now.Add(time.Duration(rand.Int63n(int64(startJitter))))
        }
        for _, queueCollections := range t.processingQueueCollections {
                t.upsertPollTime(queueCollections.Level(), newPollTime)
        }

        t.shutdownWG.Add(1)
        go t.processorPump()
}

func (t *timerQueueProcessorBase) Stop() {
        if !atomic.CompareAndSwapInt32(&t.status, common.DaemonStatusStarted, common.DaemonStatusStopped) {
                return
        }

        t.logger.Info("Timer queue processor state changed", tag.LifeCycleStopping)
        defer t.logger.Info("Timer queue processor state changed", tag.LifeCycleStopped)

        t.timerGate.Close()
        close(t.shutdownCh)
        t.pollTimeLock.Lock()
        for _, timer := range t.backoffTimer {
                timer.Stop()
        }
        t.pollTimeLock.Unlock()

        if success := common.AwaitWaitGroup(&t.shutdownWG, time.Minute); !success {
                t.logger.Warn("", tag.LifeCycleStopTimedout)
        }

        t.redispatcher.Stop()
}

func (t *timerQueueProcessorBase) processorPump() {
        defer t.shutdownWG.Done()

        updateAckTimer := time.NewTimer(backoff.JitDuration(
                t.options.UpdateAckInterval(),
                t.options.UpdateAckIntervalJitterCoefficient(),
        ))
        defer updateAckTimer.Stop()

        splitQueueTimer := time.NewTimer(backoff.JitDuration(
                t.options.SplitQueueInterval(),
                t.options.SplitQueueIntervalJitterCoefficient(),
        ))
        defer splitQueueTimer.Stop()

processorPumpLoop:
        for {
                select {
                case <-t.shutdownCh:
                        break processorPumpLoop
                case <-t.timerGate.FireChan():
                        maxRedispatchQueueSize := t.options.MaxRedispatchQueueSize()
                        if t.redispatcher.Size() > maxRedispatchQueueSize {
                                t.redispatcher.Redispatch(maxRedispatchQueueSize)
                                if t.redispatcher.Size() > maxRedispatchQueueSize {
                                        // if redispatcher still has a large number of tasks
                                        // this only happens when system is under very high load
                                        // we should backoff here instead of keeping submitting tasks to task processor
                                        // don't call t.timerGate.Update(time.Now() + loadQueueTaskThrottleRetryDelay) as the time in
                                        // standby timer processor is not real time and is managed separately
                                        time.Sleep(backoff.JitDuration(
                                                t.options.PollBackoffInterval(),
                                                t.options.PollBackoffIntervalJitterCoefficient(),
                                        ))
                                }
                                t.timerGate.Update(time.Time{})
                                continue processorPumpLoop
                        }

                        t.pollTimeLock.Lock()
                        levels := make(map[int]struct{})
                        now := t.shard.GetCurrentTime(t.clusterName)
                        for level, pollTime := range t.nextPollTime {
                                if !now.Before(pollTime) {
                                        levels[level] = struct{}{}
                                        delete(t.nextPollTime, level)
                                } else {
                                        t.timerGate.Update(pollTime)
                                }
                        }
                        t.pollTimeLock.Unlock()

                        t.processQueueCollections(levels)
                case <-updateAckTimer.C:
                        processFinished, _, err := t.updateAckLevel()
                        if err == shard.ErrShardClosed || (err == nil && processFinished) {
                                go t.Stop()
                                break processorPumpLoop
                        }
                        updateAckTimer.Reset(backoff.JitDuration(
                                t.options.UpdateAckInterval(),
                                t.options.UpdateAckIntervalJitterCoefficient(),
                        ))
                case <-t.newTimerCh:
                        t.newTimeLock.Lock()
                        newTime := t.newTime
                        t.newTime = time.Time{}
                        t.newTimeLock.Unlock()

                        // New Timer has arrived.
                        t.metricsScope.IncCounter(metrics.NewTimerNotifyCounter)
                        // notify all queue collections as they are waiting for the notification when there's
                        // no more task to process. For non-default queue, we choose to do periodic polling
                        // in the future, then we don't need to notify them.
                        for _, queueCollection := range t.processingQueueCollections {
                                t.upsertPollTime(queueCollection.Level(), newTime)
                        }
                case <-splitQueueTimer.C:
                        t.splitQueue()
                        splitQueueTimer.Reset(backoff.JitDuration(
                                t.options.SplitQueueInterval(),
                                t.options.SplitQueueIntervalJitterCoefficient(),
                        ))
                case notification := <-t.actionNotifyCh:
                        t.handleActionNotification(notification)
                }
        }
}

func (t *timerQueueProcessorBase) processQueueCollections(levels map[int]struct{}) {
        for _, queueCollection := range t.processingQueueCollections {
                level := queueCollection.Level()
                if _, ok := levels[level]; !ok {
                        continue
                }

                activeQueue := queueCollection.ActiveQueue()
                if activeQueue == nil {
                        // process for this queue collection has finished
                        // it's possible that new queue will be added to this collection later though,
                        // pollTime will be updated after split/merge
                        continue
                }

                t.upsertPollTime(level, t.shard.GetCurrentTime(t.clusterName).Add(backoff.JitDuration(
                        t.options.MaxPollInterval(),
                        t.options.MaxPollIntervalJitterCoefficient(),
                )))

                var nextPageToken []byte
                readLevel := activeQueue.State().ReadLevel()
                maxReadLevel := minTaskKey(activeQueue.State().MaxLevel(), t.updateMaxReadLevel())
                domainFilter := activeQueue.State().DomainFilter()

                if progress, ok := t.processingQueueReadProgress[level]; ok {
                        if progress.currentQueue == activeQueue {
                                readLevel = progress.readLevel
                                maxReadLevel = progress.maxReadLevel
                                nextPageToken = progress.nextPageToken
                        }
                        delete(t.processingQueueReadProgress, level)
                }

                if !readLevel.Less(maxReadLevel) {
                        // notify timer gate about the min time
                        t.upsertPollTime(level, readLevel.(timerTaskKey).visibilityTimestamp)
                        continue
                }

                ctx, cancel := context.WithTimeout(context.Background(), loadQueueTaskThrottleRetryDelay)
                if err := t.rateLimiter.Wait(ctx); err != nil {
                        cancel()
                        if level == defaultProcessingQueueLevel {
                                t.upsertPollTime(level, time.Time{})
                        } else {
                                t.setupBackoffTimer(level)
                        }
                        continue
                }
                cancel()

                timerTaskInfos, lookAheadTask, nextPageToken, err := t.readAndFilterTasks(readLevel, maxReadLevel, nextPageToken)
                if err != nil {
                        t.logger.Error("Processor unable to retrieve tasks", tag.Error(err))
                        t.upsertPollTime(level, time.Time{}) // re-enqueue the event
                        continue
                }

                tasks := make(map[task.Key]task.Task)
                taskChFull := false
                for _, taskInfo := range timerTaskInfos {
                        if !domainFilter.Filter(taskInfo.GetDomainID()) {
                                continue
                        }

                        task := t.taskInitializer(taskInfo)
                        tasks[newTimerTaskKey(taskInfo.GetVisibilityTimestamp(), taskInfo.GetTaskID())] = task
                        submitted, err := t.submitTask(task)
                        if err != nil {
                                // only err here is due to the fact that processor has been shutdown
                                // return instead of continue
                                return
                        }
                        taskChFull = taskChFull || !submitted
                }

                var newReadLevel task.Key
                if len(nextPageToken) == 0 {
                        newReadLevel = maxReadLevel
                        if lookAheadTask != nil {
                                // lookAheadTask may exist only when nextPageToken is empty
                                // notice that lookAheadTask.VisibilityTimestamp may be larger than shard max read level,
                                // which means new tasks can be generated before that timestamp. This issue is solved by
                                // upsertPollTime whenever there are new tasks
                                t.upsertPollTime(level, lookAheadTask.VisibilityTimestamp)
                                newReadLevel = minTaskKey(newReadLevel, newTimerTaskKey(lookAheadTask.GetVisibilityTimestamp(), 0))
                        }
                        // else we have no idea when the next poll should happen
                        // rely on notifyNewTask to trigger the next poll even for non-default queue.
                        // another option for non-default queue is that we can setup a backoff timer to check back later
                } else {
                        // more tasks should be loaded for this processing queue
                        // record the current progress and update the poll time
                        if level == defaultProcessingQueueLevel || !taskChFull {
                                t.upsertPollTime(level, time.Time{})
                        } else {
                                t.setupBackoffTimer(level)
                        }
                        t.processingQueueReadProgress[level] = timeTaskReadProgress{
                                currentQueue:  activeQueue,
                                readLevel:     readLevel,
                                maxReadLevel:  maxReadLevel,
                                nextPageToken: nextPageToken,
                        }
                        newReadLevel = newTimerTaskKey(timerTaskInfos[len(timerTaskInfos)-1].GetVisibilityTimestamp(), 0)
                }
                queueCollection.AddTasks(tasks, newReadLevel)
        }
}

func (t *timerQueueProcessorBase) splitQueue() {
        splitPolicy := t.initializeSplitPolicy(
                func(key task.Key, domainID string) task.Key {
                        return newTimerTaskKey(
                                key.(timerTaskKey).visibilityTimestamp.Add(
                                        t.options.SplitLookAheadDurationByDomainID(domainID),
                                ),
                                0,
                        )
                },
        )

        t.splitProcessingQueueCollection(splitPolicy, t.upsertPollTime)
}

func (t *timerQueueProcessorBase) handleActionNotification(notification actionNotification) {
        t.processorBase.handleActionNotification(notification, func() {
                switch notification.action.ActionType {
                case ActionTypeReset:
                        t.upsertPollTime(defaultProcessingQueueLevel, time.Time{})
                }
        })
}

func (t *timerQueueProcessorBase) readAndFilterTasks(
        readLevel task.Key,
        maxReadLevel task.Key,
        nextPageToken []byte,
) ([]*persistence.TimerTaskInfo, *persistence.TimerTaskInfo, []byte, error) {
        timerTasks, nextPageToken, err := t.getTimerTasks(readLevel, maxReadLevel, nextPageToken, t.options.BatchSize())
        if err != nil {
                return nil, nil, nil, err
        }

        var lookAheadTask *persistence.TimerTaskInfo
        filteredTasks := []*persistence.TimerTaskInfo{}

        for _, timerTask := range timerTasks {
                if !t.isProcessNow(timerTask.GetVisibilityTimestamp()) {
                        lookAheadTask = timerTask
                        nextPageToken = nil
                        break
                }
                filteredTasks = append(filteredTasks, timerTask)
        }

        if len(nextPageToken) == 0 && lookAheadTask == nil {
                // only look ahead within the processing queue boundary
                lookAheadTask, err = t.readLookAheadTask(maxReadLevel, maximumTimerTaskKey)
                if err != nil {
                        // we don't know if look ahead task exists or not, but we know if it exists,
                        // it's visibility timestamp is larger than or equal to maxReadLevel.
                        // so, create a fake look ahead task so another load can be triggered at that time.
                        lookAheadTask = &persistence.TimerTaskInfo{
                                VisibilityTimestamp: maxReadLevel.(timerTaskKey).visibilityTimestamp,
                        }
                        return filteredTasks, lookAheadTask, nil, nil
                }
        }

        return filteredTasks, lookAheadTask, nextPageToken, nil
}

func (t *timerQueueProcessorBase) readLookAheadTask(
        lookAheadStartLevel task.Key,
        lookAheadMaxLevel task.Key,
) (*persistence.TimerTaskInfo, error) {
        tasks, _, err := t.getTimerTasks(
                lookAheadStartLevel,
                lookAheadMaxLevel,
                nil,
                1,
        )
        if err != nil {
                return nil, err
        }

        if len(tasks) == 1 {
                return tasks[0], nil
        }
        return nil, nil
}

func (t *timerQueueProcessorBase) getTimerTasks(
        readLevel task.Key,
        maxReadLevel task.Key,
        nextPageToken []byte,
        batchSize int,
) ([]*persistence.TimerTaskInfo, []byte, error) {
        request := &persistence.GetTimerIndexTasksRequest{
                MinTimestamp:  readLevel.(timerTaskKey).visibilityTimestamp,
                MaxTimestamp:  maxReadLevel.(timerTaskKey).visibilityTimestamp,
                BatchSize:     batchSize,
                NextPageToken: nextPageToken,
        }

        var err error
        var response *persistence.GetTimerIndexTasksResponse
        retryCount := t.shard.GetConfig().TimerProcessorGetFailureRetryCount()
        for attempt := 0; attempt < retryCount; attempt++ {
                response, err = t.shard.GetExecutionManager().GetTimerIndexTasks(context.Background(), request)
                if err == nil {
                        return response.Timers, response.NextPageToken, nil
                }
                backoff := time.Duration(attempt * 100)
                time.Sleep(backoff * time.Millisecond)
        }
        return nil, nil, err
}

func (t *timerQueueProcessorBase) isProcessNow(
        expiryTime time.Time,
) bool {
        if expiryTime.IsZero() {
                // return true, but somewhere probably have bug creating empty timerTask.
                t.logger.Warn("Timer task has timestamp zero")
        }
        return expiryTime.UnixNano() <= t.shard.GetCurrentTime(t.clusterName).UnixNano()
}

func (t *timerQueueProcessorBase) notifyNewTimers(
        timerTasks []persistence.Task,
) {
        if len(timerTasks) == 0 {
                return
        }

        isActive := t.options.MetricScope == metrics.TimerActiveQueueProcessorScope

        minNewTime := timerTasks[0].GetVisibilityTimestamp()
        for _, timerTask := range timerTasks {
                ts := timerTask.GetVisibilityTimestamp()
                if ts.Before(minNewTime) {
                        minNewTime = ts
                }

                taskScopeIdx := task.GetTimerTaskMetricScope(
                        timerTask.GetType(),
                        isActive,
                )
                t.metricsClient.IncCounter(taskScopeIdx, metrics.NewTimerCounter)
        }

        t.notifyNewTimer(minNewTime)
}

func (t *timerQueueProcessorBase) notifyNewTimer(
        newTime time.Time,
) {
        t.newTimeLock.Lock()
        defer t.newTimeLock.Unlock()

        if t.newTime.IsZero() || newTime.Before(t.newTime) {
                t.newTime = newTime
                select {
                case t.newTimerCh <- struct{}{}:
                        // Notified about new time.
                default:
                        // Channel "full" -> drop and move on, this will happen only if service is in high load.
                }
        }
}

func (t *timerQueueProcessorBase) upsertPollTime(level int, newPollTime time.Time) {
        t.pollTimeLock.Lock()
        defer t.pollTimeLock.Unlock()

        if _, ok := t.backoffTimer[level]; ok {
                // honor existing backoff timer
                return
        }

        if currentPollTime, ok := t.nextPollTime[level]; !ok || newPollTime.Before(currentPollTime) {
                t.nextPollTime[level] = newPollTime
                t.timerGate.Update(newPollTime)
        }
}

// setupBackoffTimer will trigger a poll for the specified processing queue collection
// after a certain period of (real) time. This means for standby timer, even if the cluster time
// has not been updated, the poll will still be triggered when the timer fired. Use this function
// for delaying the load for processing queue. If a poll should be triggered immediately
// use upsertPollTime.
func (t *timerQueueProcessorBase) setupBackoffTimer(level int) {
        t.pollTimeLock.Lock()
        defer t.pollTimeLock.Unlock()

        if _, ok := t.backoffTimer[level]; ok {
                // honor existing backoff timer
                return
        }

        t.metricsScope.IncCounter(metrics.ProcessingQueueThrottledCounter)
        t.logger.Info("Throttled processing queue", tag.QueueLevel(level))
        backoffDuration := backoff.JitDuration(
                t.options.PollBackoffInterval(),
                t.options.PollBackoffIntervalJitterCoefficient(),
        )
        t.backoffTimer[level] = time.AfterFunc(backoffDuration, func() {
                select {
                case <-t.shutdownCh:
                        return
                default:
                }

                t.pollTimeLock.Lock()
                defer t.pollTimeLock.Unlock()

                t.nextPollTime[level] = time.Time{}
                t.timerGate.Update(time.Time{})
                delete(t.backoffTimer, level)
        })
}

func newTimerTaskKey(
        visibilityTimestamp time.Time,
        taskID int64,
) task.Key {
        return timerTaskKey{
                visibilityTimestamp: visibilityTimestamp,
                taskID:              taskID,
        }
}

func (k timerTaskKey) Less(
        key task.Key,
) bool {
        timerKey := key.(timerTaskKey)
        if k.visibilityTimestamp.Equal(timerKey.visibilityTimestamp) {
                return k.taskID < timerKey.taskID
        }
        return k.visibilityTimestamp.Before(timerKey.visibilityTimestamp)
}

func (k timerTaskKey) String() string {
        return fmt.Sprintf("{visibilityTimestamp: %v, taskID: %v}", k.visibilityTimestamp, k.taskID)
}

func newTimerQueueProcessorOptions(
        config *config.Config,
        isActive bool,
        isFailover bool,
) *queueProcessorOptions {
        options := &queueProcessorOptions{
                BatchSize:                            config.TimerTaskBatchSize,
                DeleteBatchSize:                      config.TimerTaskDeleteBatchSize,
                MaxPollRPS:                           config.TimerProcessorMaxPollRPS,
                MaxPollInterval:                      config.TimerProcessorMaxPollInterval,
                MaxPollIntervalJitterCoefficient:     config.TimerProcessorMaxPollIntervalJitterCoefficient,
                UpdateAckInterval:                    config.TimerProcessorUpdateAckInterval,
                UpdateAckIntervalJitterCoefficient:   config.TimerProcessorUpdateAckIntervalJitterCoefficient,
                RedispatchIntervalJitterCoefficient:  config.TaskRedispatchIntervalJitterCoefficient,
                MaxRedispatchQueueSize:               config.TimerProcessorMaxRedispatchQueueSize,
                SplitQueueInterval:                   config.TimerProcessorSplitQueueInterval,
                SplitQueueIntervalJitterCoefficient:  config.TimerProcessorSplitQueueIntervalJitterCoefficient,
                PollBackoffInterval:                  config.QueueProcessorPollBackoffInterval,
                PollBackoffIntervalJitterCoefficient: config.QueueProcessorPollBackoffIntervalJitterCoefficient,
        }

        if isFailover {
                // disable queue split for failover processor
                options.EnableSplit = dynamicconfig.GetBoolPropertyFn(false)

                // disable persist and load processing queue states for failover processor as it will never be split
                options.EnablePersistQueueStates = dynamicconfig.GetBoolPropertyFn(false)
                options.EnableLoadQueueStates = dynamicconfig.GetBoolPropertyFn(false)

                options.MaxStartJitterInterval = config.TimerProcessorFailoverMaxStartJitterInterval
        } else {
                options.EnableSplit = config.QueueProcessorEnableSplit
                options.SplitMaxLevel = config.QueueProcessorSplitMaxLevel
                options.EnableRandomSplitByDomainID = config.QueueProcessorEnableRandomSplitByDomainID
                options.RandomSplitProbability = config.QueueProcessorRandomSplitProbability
                options.EnablePendingTaskSplitByDomainID = config.QueueProcessorEnablePendingTaskSplitByDomainID
                options.PendingTaskSplitThreshold = config.QueueProcessorPendingTaskSplitThreshold
                options.EnableStuckTaskSplitByDomainID = config.QueueProcessorEnableStuckTaskSplitByDomainID
                options.StuckTaskSplitThreshold = config.QueueProcessorStuckTaskSplitThreshold
                options.SplitLookAheadDurationByDomainID = config.QueueProcessorSplitLookAheadDurationByDomainID

                options.EnablePersistQueueStates = config.QueueProcessorEnablePersistQueueStates
                options.EnableLoadQueueStates = config.QueueProcessorEnableLoadQueueStates

                options.MaxStartJitterInterval = dynamicconfig.GetDurationPropertyFn(0)
        }

        if isActive {
                options.MetricScope = metrics.TimerActiveQueueProcessorScope
                options.RedispatchInterval = config.ActiveTaskRedispatchInterval
        } else {
                options.MetricScope = metrics.TimerStandbyQueueProcessorScope
                options.RedispatchInterval = config.StandbyTaskRedispatchInterval
        }

        return options
}

uber / cadence / 01875e2f-959c-4c4d-87af-1d7805759bcc

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous