99
syncStateRetriever SyncStateRetriever,
100
logger log.Logger,
102
>
103
>
currentClusterName := shardContext.GetClusterMetadata().GetCurrentClusterName()
104
>
config := shardContext.GetConfig()
105
>
106
>
retryPolicy := backoff.NewExponentialRetryPolicy(200 * time.Millisecond).
107
>
WithMaximumAttempts(5).
108
>
WithBackoffCoefficient(1)
109
>
110
>
return &ackMgrImpl{
111
>
currentClusterName: currentClusterName,
112
>
shardContext: shardContext,
113
>
config: shardContext.GetConfig(),
114
>
workflowCache: workflowCache,
115
>
eventBlobCache: eventBlobCache,
116
>
replicationProgressCache: replicationProgressCache,
117
>
executionMgr: executionMgr,
118
>
metricsHandler: shardContext.GetMetricsHandler().WithTags(metrics.OperationTag(metrics.ReplicatorQueueProcessorScope)),
119
>
logger: log.With(logger, tag.ComponentReplicatorQueue),
120
>
retryPolicy: retryPolicy,
121
>
namespaceRegistry: shardContext.GetNamespaceRegistry(),
122
>
syncVersionedTransitionTaskConverter: newSyncVersionedTransitionTaskConverter(shardContext, workflowCache, eventBlobCache, replicationProgressCache, executionMgr, syncStateRetriever, logger),
123
>
pageSize: config.ReplicatorProcessorFetchTasksBatchSize,
124
>
maxSkipTaskCount: config.ReplicatorProcessorMaxSkipTaskCount,
125
>
126
>
maxTaskID: nil,
127
>
sanityCheckTime: time.Time{},
128
>
129
>
subscribers: make(map[string]channelMetadata),
130
>
}
131
>
}
132
133
func (p *ackMgrImpl) NotifyNewTasks(