128
}
129
130
>
func (tr *fairTaskReader) completeTask(task *internalTask, res taskResponse) {
fair_task_reader.go
131
>
recordDroppedTask(tr.backlogMgr.metricsHandler, res.dropReason)
132
>
133
>
tr.lock.Lock()
134
>
135
>
// We might have a race where mergeTasks tries to read a task from matcher (because new tasks
136
>
// came in under it), but it had already been matched and removed. In that case the
137
>
// removeFromMatcher will be a no-op, and we'll eventually end up here. We can tell because
138
>
// the task won't be present in outstandingTasks.
139
>
//
140
>
// We can't ack the task, so we'll eventually read it again and then discover that it's a
141
>
// duplicate when we try to RecordTaskStarted.
142
>
if task, found := tr.outstandingTasks.Get(fairLevelFromAllocatedTask(task.event.AllocatedTaskInfo)); !found {
143
metrics.TaskCompletedMissing.With(tr.backlogMgr.metricsHandler).Record(1)
144
tr.lock.Unlock()
145
return
146
>
} else if _, ok := task.(*internalTask); !softassert.That(tr.logger, ok, "completed task was already acked") {
fair_task_reader.go
147
tr.lock.Unlock()
148
return