| 361 | } |
| 362 | |
| 363 | func (j *JobV2) runOnce() { |
| 364 | defer j.ResetAllOnce() |
| 365 | |
| 366 | j.applyPendingVnodeUpdate() |
| 367 | |
| 368 | curTime := time.Now() |
| 369 | sinceLastRun := calcSinceLastRun(curTime, j.prevRun) |
| 370 | j.prevRun = curTime |
| 371 | |
| 372 | prepared, ok := j.collectAndEmit(sinceLastRun) |
| 373 | if ok && !j.panicked.Load() { |
| 374 | if err := j.finishPreparedEmission(prepared); err != nil { |
| 375 | j.Warningf("finalize emission failed: %v", err) |
| 376 | ok = false |
| 377 | } |
| 378 | } |
| 379 | if ok { |
| 380 | j.retries.Store(0) |
| 381 | } else { |
| 382 | j.retries.Add(1) |
| 383 | } |
| 384 | j.buf.Reset() |
| 385 | } |
| 386 | |
| 387 | func (j *JobV2) applyPendingVnodeUpdate() { |
| 388 | select { |