313
ctx context.Context,
314
replicationTask *replicationspb.ReplicationTask,
316
>
_ = p.rateLimiter.Wait(ctx)
317
>
318
>
operationTagValue := p.getOperationTagValue(replicationTask)
319
>
320
>
operation := func() error {
321
>
var err error
322
>
// Tests use this hook to pause replication tasks before they are applied.
323
>
if hook, ok := testhooks.Get(p.testHooks, testhooks.HistoryReplicationTaskInterceptor, testhooks.GlobalScope); ok {
324
err = hook(replicationTask, func() error {
325
return p.replicationTaskExecutor.Execute(ctx, replicationTask, false)
326
})
328
>
err = p.replicationTaskExecutor.Execute(ctx, replicationTask, false)
329
>
}
330
p.emitTaskMetrics(operationTagValue, err)
331
return err
332
}
333
335
>
defer func() {
336
>
if panicErr != nil {
337
retErr = panicErr
338
p.emitTaskMetrics(operationTagValue, panicErr)