348
ctx context.Context,
349
request *persistence.PutReplicationTaskToDLQRequest,
351
>
_ = p.rateLimiter.Wait(ctx)
352
>
353
>
// Route the DLQ entry to the shard that owns the workflow on this (target) cluster,
354
>
// not the processor's shard. In cross-shard replication (source shard count < target
355
>
// shard count), the processor shard does not always equal the owner shard. Using
356
>
// p.config.GetShardID matches what handler.GetDLQReplicationMessages uses to dispatch,
357
>
// so the operator-facing "DLQ for workflow W is at GetShardID(W)" model holds.
358
>
targetShardID := p.config.GetShardID(
359
>
namespace.ID(request.TaskInfo.GetNamespaceId()),
360
>
request.TaskInfo.GetWorkflowId(),
361
>
)
362
>
363
>
p.logger.Info("enqueue replication task to DLQ",
364
>
tag.TargetShardID(targetShardID),
365
>
tag.SourceShardID(p.sourceShardID),
366
>
tag.WorkflowNamespaceID(request.TaskInfo.GetNamespaceId()),
367
>
tag.WorkflowID(request.TaskInfo.GetWorkflowId()),
368
>
tag.WorkflowRunID(request.TaskInfo.GetRunId()),
369
>
tag.TaskID(request.TaskInfo.GetTaskId()),
370
>
)
371
>
metrics.ReplicationDLQMaxLevelGauge.With(p.metricsHandler).Record(
372
>
float64(request.TaskInfo.GetTaskId()),
373
>
metrics.OperationTag(metrics.ReplicationDLQStatsScope),
374
>
metrics.TargetClusterTag(p.sourceCluster),
375
>
metrics.InstanceTag(convert.Int32ToString(targetShardID)))
376
>
// The following is guaranteed to success or retry forever until processor is shutdown.
377
>
return backoff.ThrottleRetry(func() error {
378
>
err := writeTaskToDLQ(ctx, p.dlqWriter, p.sourceShardID, request.SourceClusterName, targetShardID, request.TaskInfo)
379
>
if err != nil {
380
p.logger.Error("failed to enqueue replication task to DLQ", tag.Error(err))
381
metrics.ReplicationDLQFailed.With(p.metricsHandler).Record(