Add a processor for the archival queue #3560
Conversation
MichaelSnowden
force-pushed
the
da-q-processors
branch
from
859f3d9 to
86440bb
Compare
MichaelSnowden
force-pushed
the
da-q-processors
branch
from
86440bb to
f826cf1
Compare
| @@ -506,6 +508,7 @@ func NewConfig(dc *dynamicconfig.Collection, numberOfShards int32, isAdvancedVis | |||
|
|
|||
| // Archival related | |||
| ArchivalTaskBatchSize: dc.GetIntProperty(dynamicconfig.ArchivalTaskBatchSize, 100), | |||
| ArchivalTaskMaxRetryCount: dc.GetIntProperty(dynamicconfig.ArchivalTaskMaxRetryCount, math.MaxInt32), | |||
There was a problem hiding this comment.
Where is this use? and the maxInt32 doesn't make sense
There was a problem hiding this comment.
I use it in a downstream PR, but I'll remove it from this one.
| shardContext shard.Context | ||
| workflowCache workflow.Cache | ||
| logger log.Logger | ||
| metricsClient metrics.MetricsHandler |
| metricsClient metrics.MetricsHandler | ||
| } | ||
|
|
||
| func newArchivalQueueTaskExecutor(archiver archival.Archiver, shardContext shard.Context, workflowCache workflow.Cache, |
There was a problem hiding this comment.
nit: one argument per line could be cleaner
| func (e *archivalQueueTaskExecutor) Execute(ctx context.Context, executable queues.Executable) (tags []metrics.Tag, | ||
| isActive bool, err error) { |
There was a problem hiding this comment.
nit:
func (e *archivalQueueTaskExecutor) Execute(
ctx context.Context,
executable queues.Executable,
) (tags []metrics.Tag,isActive bool, err error) {
| if err != nil { | ||
| return err | ||
| } | ||
| if mutableState == nil || mutableState.IsWorkflowExecutionRunning() { |
There was a problem hiding this comment.
If the mutable state is nil, I think we should return nil to ack this task. If the workflow is deleted manually prior to archival.
| if retention == 0 { | ||
| retention = 7 * 24 * time.Hour | ||
| } | ||
| deleteTime := closeTime.Add(retention) |
There was a problem hiding this comment.
Done in #3565
| } | ||
| retention := namespaceEntry.Retention() | ||
| if retention == 0 { | ||
| retention = 7 * 24 * time.Hour |
| StartTime: *startTime, | ||
| ExecutionTime: *executionTime, | ||
| CloseTime: *closeTime, |
There was a problem hiding this comment.
Can the archival takes in an optional of those fields so you don't need those sanity check? And if those sanity check need to be there, I think it belongs to archiver.
There was a problem hiding this comment.
Makes sense, fixed
|
Moved to #3663 |
What changed?
Added a processor which executes tasks on the archival queue. A followup PR will actually enable the queue factory.
Why?
I made these changes to enable the archival queue.
How did you test it?
I tested this using unit tests because it's not actually hooked into prod yet.
Potential risks
No--it's not wired into main yet.
Is hotfix candidate?
No.