53
matchingClient matchingservice.MatchingServiceClient,
54
namespaceRegistry namespace.Registry,
56
>
defaultRetryPolicy := backoff.NewExponentialRetryPolicy(taskProcessorErrorRetryWait).
57
>
WithBackoffCoefficient(taskProcessorErrorRetryBackoffCoefficient).
58
>
WithMaximumAttempts(taskProcessorErrorRetryMaxAttampts)
59
>
60
>
namespaceTaskRetryPolicy := backoff.NewExponentialRetryPolicy(taskProcessorErrorRetryWait).
61
>
WithBackoffCoefficient(taskProcessorErrorRetryBackoffCoefficient).
62
>
WithMaximumAttempts(namespaceTaskRetryMaxAttempts)
63
>
64
>
// Namespace tasks get more retry attempts because they're gated on the
65
>
// cluster-global namespace metadata CAS; other task types stay on the
66
>
// shorter default to keep the single-threaded loop responsive.
67
>
retryPolicyForTask := func(task *replicationspb.ReplicationTask) backoff.RetryPolicy {
68
>
switch task.TaskType {
69
>
case enumsspb.REPLICATION_TASK_TYPE_NAMESPACE_TASK:
70
>
return namespaceTaskRetryPolicy
71
>
default:
72
>
return defaultRetryPolicy
73
}
74
}
75
77
>
hostInfo: hostInfo,
78
>
serviceResolver: serviceResolver,
79
>
status: common.DaemonStatusInitialized,
80
>
currentCluster: currentCluster,
81
>
sourceCluster: sourceCluster,
82
>
logger: logger,
83
>
remotePeer: remotePeer,
84
>
namespaceTaskExecutor: namespaceTaskExecutor,
85
>
customTaskHandler: customTaskHandler,
86
>
metricsHandler: metricsHandler.WithTags(metrics.OperationTag(metrics.NamespaceReplicationTaskScope)),
87
>
retryPolicyForTask: retryPolicyForTask,
88
>
lastProcessedMessageID: -1,
89
>
lastRetrievedMessageID: -1,
90
>
done: make(chan struct{}),
91
>
namespaceReplicationQueue: namespaceReplicationQueue,
92
>
matchingClient: matchingClient,
93
>
namespaceRegistry: namespaceRegistry,
94
>
}
95
}
96