damccorm opened a new pull request, #41146: URL: https://github.com/apache/beam/pull/41146
## Summary Fixes a `prism error: negative watermark hold count -1` panic in `(*holdTracker).Drop` observed in `GroupIntoBatchesTest.testInGlobalWindowBatchSizeByteSizeFn` during `PreCommit Java PVR Prism Loopback` ([run 42742771136](https://github.com/apache/beam/actions/runs/42742771136/job/129587622470)). ### Root Cause 1. **Stale overwritten event-time timer matching when only `holdTimestamp` changes**: - In `GroupIntoBatchesDoFn`, the event-time watermark hold timer (`ts-watermarkHold`) always sets its firing timestamp to `window.maxTimestamp().plus(allowedLateness)` (`GlobalWindow.MaxTimestamp`) while updating its `holdTimestamp` as batches flush and new elements arrive. - When `(*statefulStageKind).addPending` overwrites an existing event-time timer, it drops `lastSet.hold` from `ss.watermarkHolds`, adds the new `e.holdTimestamp`, and updates `dnt.timers[tk] = {firing: e.timestamp, hold: e.holdTimestamp}`. However, the superseded timer `element` remains in the per-key binary heap `dnt.elements`. - When `(*statefulStageKind).buildEventTimeBundle` later pops from `dnt.elements`, it only checked `e.timestamp != lastSet.firing` and did not check `e.holdTimestamp != lastSet.hold` (or `e.sequence < 0`). When a superseded timer with the same firing timestamp (`MaxTimestamp`) but an older `holdTimestamp` (`4000`) was popped before the current timer (`9000`), `buildEventTimeBundle` treated it as the active timer, recorded `holdsInBundle[4000]++`, and deleted `dnt.timers[tk]`. Upon bundle completion, `PersistBundle` dropped `4000` a second time, panicking with `negative watermark hold count -1 for time 4000`. 2. **Cleared processing-time timers (`sequence < 0`) scheduled to fire immediately**: - `decodeTimerIter` decodes a cleared timer (`clear = true`, such as `bufferingTimer.clear()` in `GroupIntoBatchesDoFn`) with `sequence = -1, timestamp = 0, holdTimestamp = 0`. - Unlike event-time timers, `triageTimers` and `(*timerHandler).Persist` did not check `sequence < 0` for processing-time timers, rebasing `0` to `emNow` and scheduling the cleared timer to fire immediately with `holdTimestamp = 0`. ### Fix - In `(*statefulStageKind).buildEventTimeBundle`, check `e.sequence < 0 || e.timestamp != lastSet.firing || e.holdTimestamp != lastSet.hold` before firing a timer element. - In `triageTimers` and `(*timerHandler).Persist`, handle cleared processing-time timers (`sequence < 0`) by removing any existing scheduled timer and decrementing its hold count without scheduling a firing. - Add unit tests in `elementmanager_test.go` and `timers_test.go`. ## Verification - Triggered `PreCommit Java PVR Prism Loopback` workflow run on branch: https://github.com/apache/beam/actions/runs/42750759120 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
