1088
// WaitCatchup waits for the CatchupCluster to catch necessary data from the current cluster,
1089
// ensuring it has caught up to the TargetCluster's ack level for the specified namespace.
1090
>
func (a *activities) WaitCatchup(ctx context.Context, params CatchUpParams) error {
activities.go
1091
>
ctx = headers.SetCallerInfo(ctx, headers.NewCallerInfo(params.Namespace, headers.CallerTypeAPI, ""))
1092
>
1093
>
descResp, err := a.frontendClient.DescribeNamespace(ctx, &workflowservice.DescribeNamespaceRequest{
1094
>
Namespace: params.Namespace,
1095
>
})
1096
>
if err != nil {
1097
return err
1098
}
1099
1101
>
if targetCluster == "" {
1102
>
targetCluster = descResp.ReplicationConfig.GetActiveClusterName()
1103
>
}
1104
1106
>
Namespace: params.Namespace,
1107
>
CatchupCluster: params.CatchupCluster,
1108
>
TargetCluster: targetCluster,
1109
>
}
1110
>
1111
>
targetAckIDOnShard, err := a.getTargetClusterReplicationStatus(ctx, waitCatchupRequest)
1112
>
if err != nil {
1113
return err
1114
}
1115
1117
>
done, err := a.checkReplicationOnRemoteCluster(ctx, waitCatchupRequest, targetAckIDOnShard)
1118
>
if err != nil {
1119
return err
1120
}
1122
>
return nil
1123
>
}
1124
1125
// keep waiting and check again