separate timer queue ack manager in separate file, add functionality … #630
Conversation
wxing1292
force-pushed
the
timer-Q-ack
branch
2 times, most recently
from
1e70506
to
0ef2268
Compare
service/history/timerQueueAckMgr.go
Outdated
|
|
||
| sync.Mutex | ||
| // outstanding tasks, key cluster name, value map of outstanding timer task -> finished (true) | ||
| clusterOutstandingTasks map[string]map[TimerSequenceID]bool |
There was a problem hiding this comment.
I think we should keep this component simple by not making it aware of all clusters. We should have a separate instance of TimerQueueProcessor with its own AckManager per cluster.
| @@ -58,8 +58,8 @@ type ( | |||
| GetConfig() *Config | |||
| GetLogger() bark.Logger | |||
| GetMetricsClient() metrics.Client | |||
| GetTimerAckLevel() time.Time | |||
| UpdateTimerAckLevel(ackLevel time.Time) error | |||
| GetTimerAckLevel(cluster string) time.Time | |||
There was a problem hiding this comment.
Now we are going to have too many components managing ack levels separately. Currently each update is a write to cassandra, this is not going to scale once cross dc support is in production. Let's put a TODO to accept updates from individual components and have a separate job which updates all the ack levels in a single write for the entire shard.
service/history/shardContext.go
Outdated
| @@ -121,6 +121,13 @@ func (s *shardContextImpl) GetTransferAckLevel() int64 { | |||
| s.RLock() | |||
| defer s.RUnlock() | |||
|
|
|||
| // TODO change make this cluster input parameter | |||
| cluster := s.GetService().GetClusterMetadata().GetCurrentClusterName() | |||
| // if can find corresponding ack level in the cluster to timer ack level map | |||
| @@ -169,6 +156,12 @@ func (t *timerQueueProcessorImpl) NotifyNewTimers(timerTasks []persistence.Task) | |||
| } | |||
| } | |||
|
|
|||
| // NotifyNewTimerEvents - Notify the processor about the new standby history events arrival. | |||
| // This should be called each time new timer events arrives, otherwise timers maybe fired unexpected. | |||
| func (t *timerQueueProcessorImpl) NotifyNewTimerEvents() { | |||
There was a problem hiding this comment.
I think we should have the same API for notifying new timers whether it is for active processor or standby processor. Basically we should have a baseTimerQueueProcessor which has common logic and then activeTimerQueueProcessor and passiveTimerQueueProcessor with their own handling of TimerTasks. But the way you notify both should be exactly the same.
There was a problem hiding this comment.
this will be done in the next PR
…to timer queue processor to be cluster aware rename SequenceID -> TimerSequenceID
…to timer queue processor to be cluster aware
rename SequenceID -> TimerSequenceID
Standby timer processing logic && domain failover timer ack manager scan DB for domain specific timer in next PR.
Partially solve #565