70
exclusiveHighWatermarkInfo WatermarkInfo,
71
tasks ...TrackableExecutableTask,
73
>
filteredTasks := make([]TrackableExecutableTask, 0, len(tasks))
74
>
75
>
t.Lock()
76
>
defer t.Unlock()
77
>
78
>
// need to assume source side send replication tasks in order
79
>
if t.exclusiveHighWatermarkInfo != nil && exclusiveHighWatermarkInfo.Watermark <= t.exclusiveHighWatermarkInfo.Watermark {
80
return filteredTasks
81
}
82
84
>
if item := t.taskQueue.Back(); item != nil {
85
lastTaskID = item.Value.(TrackableExecutableTask).TaskID()
86
}
88
>
for _, task := range tasks {
89
if lastTaskID >= task.TaskID() {
90
// need to assume source side send replication tasks in order