255
return 0, nil
256
}
257
>
highPriorityFlowControlInfo := r.flowController.GetFlowControlInfo(enumsspb.TASK_PRIORITY_HIGH)
stream_receiver.go
258
>
if highPriorityFlowControlInfo.Command == enumsspb.REPLICATION_FLOW_CONTROL_COMMAND_PAUSE {
259
r.logger.Warn(fmt.Sprintf("pausing High Priority Tasks: %s", highPriorityFlowControlInfo.Cause))
260
}
262
>
InclusiveLowWatermark: highPriorityWaterMarkInfo.Watermark,
263
>
InclusiveLowWatermarkTime: timestamppb.New(highPriorityWaterMarkInfo.Timestamp),
264
>
FlowControlCommand: highPriorityFlowControlInfo.Command,
265
>
}
266
>
lowPriorityFlowControlInfo := r.flowController.GetFlowControlInfo(enumsspb.TASK_PRIORITY_LOW)
267
>
if lowPriorityFlowControlInfo.Command == enumsspb.REPLICATION_FLOW_CONTROL_COMMAND_PAUSE {
268
>
r.logger.Warn(fmt.Sprintf("pausing Low Priority Tasks: %s", lowPriorityFlowControlInfo.Cause))
269
>
}
270
>
lowPriorityWatermark = &replicationspb.ReplicationState{
271
>
InclusiveLowWatermark: lowPriorityWaterMarkInfo.Watermark,
272
>
InclusiveLowWatermarkTime: timestamppb.New(lowPriorityWaterMarkInfo.Timestamp),
273
>
FlowControlCommand: lowPriorityFlowControlInfo.Command,
274
>
}
275
>
if highPriorityWaterMarkInfo.Watermark <= lowPriorityWaterMarkInfo.Watermark {
276
>
inclusiveLowWaterMark = highPriorityWaterMarkInfo.Watermark
277
>
inclusiveLowWaterMarkTime = highPriorityWaterMarkInfo.Timestamp
278
>
} else {
279
inclusiveLowWaterMark = lowPriorityWaterMarkInfo.Watermark
280
inclusiveLowWaterMarkTime = lowPriorityWaterMarkInfo.Timestamp