Unit tests for timer/transfer queue processor pump loops

common/util.go

@@ -870,9 +870,7 @@ func ConvertIntMapToDynamicConfigMapProperty(
 
 // ConvertDynamicConfigMapPropertyToIntMap convert a map property from dynamic config to a map
 // whose type for both key and value are int
-func ConvertDynamicConfigMapPropertyToIntMap(
-	dcValue map[string]interface{},
-) (map[int]int, error) {
+func ConvertDynamicConfigMapPropertyToIntMap(dcValue map[string]interface{}) (map[int]int, error) {
 	intMap := make(map[int]int)
 	for key, value := range dcValue {
 		intKey, err := strconv.Atoi(strings.TrimSpace(key))

service/history/config/config.go

@@ -601,6 +601,17 @@ func NewForTestByShardNumber(shardNumber int) *Config {
 	panicIfErr(inMem.UpdateValue(dynamicconfig.NormalDecisionScheduleToStartMaxAttempts, 3))
 	panicIfErr(inMem.UpdateValue(dynamicconfig.EnablePendingActivityValidation, true))
 	panicIfErr(inMem.UpdateValue(dynamicconfig.QueueProcessorEnableGracefulSyncShutdown, true))
+	panicIfErr(inMem.UpdateValue(dynamicconfig.QueueProcessorSplitMaxLevel, 2))
+	panicIfErr(inMem.UpdateValue(dynamicconfig.QueueProcessorPendingTaskSplitThreshold, map[string]interface{}{
+		"0": 1000,
+		"1": 5000,
+	}))
+	panicIfErr(inMem.UpdateValue(dynamicconfig.QueueProcessorStuckTaskSplitThreshold, map[string]interface{}{
+		"0": 10,
+		"1": 50,
+	}))
+	panicIfErr(inMem.UpdateValue(dynamicconfig.QueueProcessorRandomSplitProbability, 0.5))
+
 	dc := dynamicconfig.NewCollection(inMem, log.NewNoop())
 	config := New(dc, shardNumber, 1024*1024, config.StoreTypeCassandra, false, "")
 	// reduce the duration of long poll to increase test speed
@@ -615,6 +626,10 @@ func NewForTestByShardNumber(shardNumber int) *Config {
 	config.NormalDecisionScheduleToStartMaxAttempts = dc.GetIntPropertyFilteredByDomain(dynamicconfig.NormalDecisionScheduleToStartMaxAttempts)
 	config.PendingActivityValidationEnabled = dc.GetBoolProperty(dynamicconfig.EnablePendingActivityValidation)
 	config.QueueProcessorEnableGracefulSyncShutdown = dc.GetBoolProperty(dynamicconfig.QueueProcessorEnableGracefulSyncShutdown)
+	config.QueueProcessorSplitMaxLevel = dc.GetIntProperty(dynamicconfig.QueueProcessorSplitMaxLevel)
+	config.QueueProcessorPendingTaskSplitThreshold = dc.GetMapProperty(dynamicconfig.QueueProcessorPendingTaskSplitThreshold)
+	config.QueueProcessorStuckTaskSplitThreshold = dc.GetMapProperty(dynamicconfig.QueueProcessorStuckTaskSplitThreshold)
+	config.QueueProcessorRandomSplitProbability = dc.GetFloat64Property(dynamicconfig.QueueProcessorRandomSplitProbability)
 	return config
 }
 

service/history/queue/processor_base_test.go

@@ -332,6 +332,59 @@ func (s *processorBaseSuite) TestUpdateAckLevel_Timer_UpdateQueueStates() {
 	s.Equal(now.Add(-5*time.Second), ackLevel.(timerTaskKey).visibilityTimestamp)
 }
 
+func (s *processorBaseSuite) TestInitializeSplitPolicy_Disabled() {
+	processorBase := s.newTestProcessorBase(
+		nil,
+		nil,
+		nil,
+		nil,
+		nil,
+	)
+
+	splitPolicy := processorBase.initializeSplitPolicy(nil)
+	s.Nil(splitPolicy, "got non-nil split policy, want nil because it's disabled")
+}
+
+func (s *processorBaseSuite) TestInitializeSplitPolicy_Enabled() {
+	processorBase := s.newTestProcessorBase(nil, nil, nil, nil, nil)
+
+	processorBase.options.EnableSplit = dynamicconfig.GetBoolPropertyFn(true)
+
+	splitPolicy := processorBase.initializeSplitPolicy(nil)
+	s.NotNil(splitPolicy, "got nil split policy, want non-nil")
+	aggPolicy, ok := splitPolicy.(*aggregatedSplitPolicy)
+	s.True(ok, "got %T, want *aggregatedSplitPolicy", splitPolicy)
+	s.Equal(3, len(aggPolicy.policies), "got %v policies, want 3: pending task policy, stuck task policy and random split policy", len(aggPolicy.policies))
+}
+
+func (s *processorBaseSuite) TestResetProcessingQueueStates() {
+	processingQueueStates := []ProcessingQueueState{
+		NewProcessingQueueState(
+			0,
+			newTransferTaskKey(0),
+			newTransferTaskKey(100),
+			NewDomainFilter(map[string]struct{}{"testDomain1": {}}, true),
+		),
+		NewProcessingQueueState(
+			1,
+			newTransferTaskKey(0),
+			newTransferTaskKey(100),
+			NewDomainFilter(map[string]struct{}{"testDomain1": {}}, false),
+		),
+		NewProcessingQueueState(
+			0,
+			newTransferTaskKey(100),
+			newTransferTaskKey(1000),
+			NewDomainFilter(map[string]struct{}{}, true),
+		),
+	}
+	processorBase := s.newTestProcessorBase(processingQueueStates, nil, nil, nil, nil)
+
+	res, err := processorBase.resetProcessingQueueStates()
+	s.NoError(err, "no error expected")
+	s.Equal(ActionTypeReset, res.ActionType, "got action type %v, want %v", res.ActionType, ActionTypeReset)
+}
+
 func (s *processorBaseSuite) newTestProcessorBase(
 	processingQueueStates []ProcessingQueueState,
 	updateMaxReadLevel updateMaxReadLevelFn,

service/history/queue/timer_queue_processor_base.go

@@ -65,20 +65,21 @@ type (
 		*processorBase
 
 		taskInitializer task.Initializer
-
-		clusterName string
-
-		pollTimeLock sync.Mutex
-		backoffTimer map[int]*time.Timer
-		nextPollTime map[int]time.Time
-		timerGate    TimerGate
+		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
+
+		updateAckLevelFn                 func() (bool, task.Key, error)
+		splitProcessingQueueCollectionFn func(splitPolicy ProcessingQueueSplitPolicy, upsertPollTimeFn func(int, time.Time))
 	}
 
 	filteredTimerTasksResponse struct {
@@ -122,9 +123,8 @@ func newTimerQueueProcessorBase(
 		queueType = task.QueueTypeStandbyTimer
 	}
 
-	return &timerQueueProcessorBase{
+	t := &timerQueueProcessorBase{
 		processorBase: processorBase,
-
 		taskInitializer: func(taskInfo task.Info) task.Task {
 			return task.NewTimerTask(
 				shard,
@@ -138,17 +138,17 @@ func newTimerQueueProcessorBase(
 				shard.GetConfig().TaskCriticalRetryCount,
 			)
 		},
-
-		clusterName: clusterName,
-
-		backoffTimer: make(map[int]*time.Timer),
-		nextPollTime: make(map[int]time.Time),
-		timerGate:    timerGate,
-
-		newTimerCh: make(chan struct{}, 1),
-
+		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),
 	}
+
+	t.updateAckLevelFn = t.updateAckLevel
+	t.splitProcessingQueueCollectionFn = t.splitProcessingQueueCollection
+	return t
 }
 
 func (t *timerQueueProcessorBase) Start() {
@@ -370,7 +370,7 @@ func (t *timerQueueProcessorBase) splitQueue(splitQueueTimer *time.Timer) {
 		},
 	)
 
-	t.splitProcessingQueueCollection(splitPolicy, t.upsertPollTime)
+	t.splitProcessingQueueCollectionFn(splitPolicy, t.upsertPollTime)
 
 	splitQueueTimer.Reset(backoff.JitDuration(
 		t.options.SplitQueueInterval(),
@@ -381,7 +381,7 @@ func (t *timerQueueProcessorBase) splitQueue(splitQueueTimer *time.Timer) {
 // handleAckLevelUpdate updates ack level and resets timer with jitter.
 // returns true if processing should be terminated
 func (t *timerQueueProcessorBase) handleAckLevelUpdate(updateAckTimer *time.Timer) bool {
-	processFinished, _, err := t.updateAckLevel()
+	processFinished, _, err := t.updateAckLevelFn()
 	if err == shard.ErrShardClosed || (err == nil && processFinished) {
 		return true
 	}

service/history/queue/timer_queue_processor_base_test.go

@@ -33,6 +33,7 @@ import (
 	"github.com/uber-go/tally"
 
 	"github.com/uber/cadence/common/cluster"
+	"github.com/uber/cadence/common/dynamicconfig"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/testlogger"
 	"github.com/uber/cadence/common/metrics"
@@ -830,6 +831,75 @@ func (s *timerQueueProcessorBaseSuite) TestProcessBatch_NoNextPage_NoLookAhead()
 	}
 }
 
+func (s *timerQueueProcessorBaseSuite) TestTimerProcessorPump_HandleAckLevelUpdate() {
+	now := time.Now()
+	queueLevel := 0
+	ackLevel := newTimerTaskKey(now.Add(50*time.Millisecond), 0)
+	maxLevel := newTimerTaskKey(now.Add(10*time.Second), 0)
+	processingQueueStates := []ProcessingQueueState{
+		NewProcessingQueueState(
+			queueLevel,
+			ackLevel,
+			maxLevel,
+			NewDomainFilter(map[string]struct{}{"testDomain1": {}}, false),
+		),
+	}
+	updateMaxReadLevel := func() task.Key {
+		return newTimerTaskKey(now, 0)
+	}
+
+	timerQueueProcessBase := s.newTestTimerQueueProcessorBase(processingQueueStates, updateMaxReadLevel, nil, nil, nil)
+	timerQueueProcessBase.options.UpdateAckInterval = dynamicconfig.GetDurationPropertyFn(1 * time.Millisecond)
+	updatedCh := make(chan struct{}, 1)
+	timerQueueProcessBase.updateAckLevelFn = func() (bool, task.Key, error) {
+		updatedCh <- struct{}{}
+		return false, nil, nil
+	}
+	timerQueueProcessBase.Start()
+	defer timerQueueProcessBase.Stop()
+
+	select {
+	case <-updatedCh:
+		return
+	case <-time.After(100 * time.Millisecond):
+		s.Fail("Ack level update not called")
+	}
+}
+
+func (s *timerQueueProcessorBaseSuite) TestTimerProcessorPump_SplitQueue() {
+	now := time.Now()
+	queueLevel := 0
+	ackLevel := newTimerTaskKey(now.Add(50*time.Millisecond), 0)
+	maxLevel := newTimerTaskKey(now.Add(10*time.Second), 0)
+	processingQueueStates := []ProcessingQueueState{
+		NewProcessingQueueState(
+			queueLevel,
+			ackLevel,
+			maxLevel,
+			NewDomainFilter(map[string]struct{}{"testDomain1": {}}, false),
+		),
+	}
+	updateMaxReadLevel := func() task.Key {
+		return newTimerTaskKey(now, 0)
+	}
+
+	timerQueueProcessBase := s.newTestTimerQueueProcessorBase(processingQueueStates, updateMaxReadLevel, nil, nil, nil)
+	timerQueueProcessBase.options.SplitQueueInterval = dynamicconfig.GetDurationPropertyFn(1 * time.Millisecond)
+	splittedCh := make(chan struct{}, 1)
+	timerQueueProcessBase.splitProcessingQueueCollectionFn = func(splitPolicy ProcessingQueueSplitPolicy, upsertPollTimeFn func(int, time.Time)) {
+		splittedCh <- struct{}{}
+	}
+	timerQueueProcessBase.Start()
+	defer timerQueueProcessBase.Stop()
+
+	select {
+	case <-splittedCh:
+		return
+	case <-time.After(100 * time.Millisecond):
+		s.Fail("splitProcessingQueueCollectionFn not called")
+	}
+}
+
 func (s *timerQueueProcessorBaseSuite) newTestTimerQueueProcessorBase(
 	processingQueueStates []ProcessingQueueState,
 	updateMaxReadLevel updateMaxReadLevelFn,

service/history/queue/transfer_queue_processor_base.go

@@ -76,6 +76,9 @@ type (
 
 		// for validating if the queue failed to load any tasks
 		validator *transferQueueValidator
+
+		processQueueCollectionsFn func()
+		updateAckLevelFn          func() (bool, task.Key, error)
 	}
 )
 
@@ -113,7 +116,6 @@ func newTransferQueueProcessorBase(
 
 	transferQueueProcessorBase := &transferQueueProcessorBase{
 		processorBase: processorBase,
-
 		taskInitializer: func(taskInfo task.Info) task.Task {
 			return task.NewTransferTask(
 				shard,
@@ -127,17 +129,17 @@ func newTransferQueueProcessorBase(
 				shard.GetConfig().TaskCriticalRetryCount,
 			)
 		},
-
-		notifyCh:  make(chan struct{}, 1),
-		processCh: make(chan struct{}, 1),
-
-		backoffTimer:  make(map[int]*time.Timer),
-		shouldProcess: make(map[int]bool),
-
+		notifyCh:         make(chan struct{}, 1),
+		processCh:        make(chan struct{}, 1),
+		backoffTimer:     make(map[int]*time.Timer),
+		shouldProcess:    make(map[int]bool),
 		lastSplitTime:    time.Time{},
 		lastMaxReadLevel: 0,
 	}
 
+	transferQueueProcessorBase.processQueueCollectionsFn = transferQueueProcessorBase.processQueueCollections
+	transferQueueProcessorBase.updateAckLevelFn = transferQueueProcessorBase.updateAckLevel
+
 	if shard.GetConfig().EnableDebugMode && options.EnableValidator() {
 		transferQueueProcessorBase.validator = newTransferQueueValidator(
 			transferQueueProcessorBase,
@@ -329,10 +331,10 @@ func (t *transferQueueProcessorBase) processorPump() {
 				default:
 				}
 			} else {
-				t.processQueueCollections()
+				t.processQueueCollectionsFn()
 			}
 		case <-updateAckTimer.C:
-			processFinished, _, err := t.updateAckLevel()
+			processFinished, _, err := t.updateAckLevelFn()
 			if err == shard.ErrShardClosed || (err == nil && processFinished) {
 				if !t.options.EnableGracefulSyncShutdown() {
 					go t.Stop()