}
defer m.Unlock()
watermark.FireTime = watermark.FireTime.Truncate(monitorWatermarkPrecision)
watermark.TaskID = 0
stats := m.readerStats[readerID]
if stats.progress.watermark.CompareTo(watermark) != 0 {
stats.progress = readerProgress{
watermark: watermark,
attempts: 1,
}
m.readerStats[readerID] = stats
return
}
stats.progress.attempts++