separate transfer queue to active and standby processor #661
Conversation
wxing1292
force-pushed
the
transfer-Q-ack-x
branch
from
c42b31d
to
0283818
Compare
separate queue ack mgr into separate file
wxing1292
force-pushed
the
transfer-Q-ack-x
branch
3 times, most recently
from
0571cf6
to
e230c92
Compare
wxing1292
force-pushed
the
transfer-Q-ack-x
branch
from
e230c92
to
d6c62b8
Compare
|
|
||
| // error which will be thrown if the timer / transfer task should be | ||
| // retries due to various of reasons | ||
| taskRetryError struct{} |
There was a problem hiding this comment.
Let's reuse the same error used by timerQueueProcessor.
| } | ||
|
|
||
| transferQueueProcessor interface { | ||
| common.Daemon | ||
| NotifyNewTask() | ||
| NotifyNewTask(clusterName string) | ||
| SetCurrentTime(clusterName string, currentTime time.Time) |
There was a problem hiding this comment.
TransferQueueProcessor does not use time at all. It uses a constant value for visibilityTime for all transfer and replicator tasks.
What's the purpose of this API?
There was a problem hiding this comment.
this is to delay the processing of the standby transfer tasks.
| @@ -118,12 +132,11 @@ type ( | |||
| } | |||
|
|
|||
| timerQueueAckMgr interface { | |||
| readRetryTimerTasks() []*persistence.TimerTaskInfo | |||
| getFinishedChan() <-chan struct{} | |||
There was a problem hiding this comment.
I guess you will remove it from this PR. I think the timer changes covers this.
There was a problem hiding this comment.
plz ignore the timer* files
| @@ -99,6 +109,8 @@ func (p *replicatorQueueProcessorImpl) Process(qTask queueTaskInfo) error { | |||
|
|
|||
| if err != nil { | |||
| p.metricsClient.IncCounter(scope, metrics.TaskRequests) | |||
There was a problem hiding this comment.
Can you change this to TaskFailure?
| @@ -99,6 +109,8 @@ func (p *replicatorQueueProcessorImpl) Process(qTask queueTaskInfo) error { | |||
|
|
|||
| if err != nil { | |||
| p.metricsClient.IncCounter(scope, metrics.TaskRequests) | |||
| } else { | |||
| p.queueAckMgr.completeTask(task.TaskID) | |||
There was a problem hiding this comment.
Why move this out of base processor?
There was a problem hiding this comment.
i believe there should be some logical separation of base processor and actual processor.
the base processor should only deal with retry, when seeing a error, and the actual task processor (the actual processing logic) should deal with errors such as entity not exists.
separate queue ack mgr into separate file
Review #648 first, since this PR contains some code from #648
solve #564
partially solve #566