{"record":{"id":"4de264ccbf402f46","repo":"apache/beam","slug":"prism-error-negative-watermark-hold-count-v-for-time-v","errorCode":null,"errorMessage":"prism error: negative watermark hold count %v for time %v","messagePattern":"prism error: negative watermark hold count (.+?) for time (.+?)","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/runners/prism/internal/engine/holds.go","lineNumber":87,"sourceCode":"\theap   mtimeHeap\n\tcounts map[mtime.Time]int\n}\n\nfunc newHoldTracker() *holdTracker {\n\treturn &holdTracker{\n\t\tcounts: map[mtime.Time]int{},\n\t}\n}\n\n// Drop the given hold count. When the count of a hold time reaches zero, it's\n// removed from the heap. Drop panics if holds become negative.\nfunc (ht *holdTracker) Drop(hold mtime.Time, v int) {\n\tn := ht.counts[hold] - v\n\tif n > 0 {\n\t\tht.counts[hold] = n\n\t\treturn\n\t} else if n < 0 {\n\t\tpanic(fmt.Sprintf(\"prism error: negative watermark hold count %v for time %v\", n, hold))\n\t}\n\tdelete(ht.counts, hold)\n\tht.heap.Remove(hold)\n}\n\n// Add a hold a number of times to heap. If the hold time isn't already present in the heap, it is added.\nfunc (ht *holdTracker) Add(hold mtime.Time, v int) {\n\t// Mark the hold in the heap.\n\tht.counts[hold] += v\n\tif len(ht.counts) != len(ht.heap) {\n\t\t// Since there's a difference, the hold should not be in the heap, so we add it.\n\t\theap.Push(&ht.heap, hold)\n\t}\n}\n\n// Min returns the earliest hold in the heap. Returns [mtime.MaxTimestamp] if the heap is empty.\nfunc (ht *holdTracker) Min() mtime.Time {\n\tminWatermarkHold := mtime.MaxTimestamp","sourceCodeStart":69,"sourceCodeEnd":105,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/runners/prism/internal/engine/holds.go#L69-L105","documentation":"holdTracker.Drop removes v counts of a watermark hold at a given time. If subtracting yields a negative count, the tracker's bookkeeping is inconsistent — more holds are being dropped than were ever added — so Prism panics to avoid silently corrupting watermark advancement. It is raised from Drop, called by addPending and splitBundle.","triggerScenarios":"Calling Drop with a count v exceeding the recorded count for that hold time, e.g. double-dropping the same hold after bundle splitting or element re-pending — an accounting bug in bundle split/hold lifecycle.","commonSituations":"Dynamic work rebalancing (bundle splitting) combined with stateful/timer holds, or a runner bug where holds are added under one time and dropped under another.","solutions":["Check whether bundle splitting (splitBundle) is double-dropping holds; disable or reduce splitting (e.g. avoid fractional split attempts) to confirm","Upgrade Beam — hold-tracking bugs around splits have been fixed in past releases","Reproduce with a minimal pipeline using timers/state and event-time holds and report it if it persists","As a local diagnostic, log every Add/Drop pair to identify the unbalanced operation"],"exampleFix":"// before: unbalanced drop on split\nht.Drop(holdTime, removedCount) // removedCount includes holds not tracked\n// after: clamp/verify before dropping\nif tracked := ht.counts[holdTime]; removedCount <= tracked {\n    ht.Drop(holdTime, removedCount)\n} else {\n    slog.Warn(\"skipping over-drop\", \"have\", tracked, \"want\", removedCount)\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"defer func() {\n    if r := recover(); r != nil {\n        if strings.Contains(fmt.Sprint(r), \"negative watermark hold count\") {\n            log.Printf(\"hold tracker invariant failure: %v\", r)\n        }\n        panic(r)\n    }\n}()","preventionTips":["Instrument Add/Drop call sites when using timers with bundle splitting","Keep Beam versions current; hold-tracking around splits has known past fixes","Reproduce with minimal timers+state pipelines before filing bugs"],"tags":["go","beam","prism","watermark","panic"],"backgroundTag":"internal-invariant-violation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}