{"record":{"id":"f7fff08d0085db22","repo":"apache/beam","slug":"source-failed-processing-timers","errorCode":null,"errorMessage":"source failed processing timers","messagePattern":"source failed processing timers","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/datasource.go","lineNumber":155,"sourceCode":"\t\t\t// Channel closed, so time to exit\n\t\t\tif !ok {\n\t\t\t\treturn nil\n\t\t\t}\n\t\t\tif len(e.Data) > 0 {\n\t\t\t\tr.Reset(e.Data)\n\t\t\t\terr = data(&bcr, e.PtransformID)\n\t\t\t}\n\t\t\tif err != nil && err != io.EOF {\n\t\t\t\treturn errors.Wrapf(err, \"source failed processing data\")\n\t\t\t}\n\t\t\t// Process any simultaneously sent timers.\n\t\t\t// If the data channel has split though\n\t\t\tif len(e.Timers) > 0 {\n\t\t\t\tr.Reset(e.Timers)\n\t\t\t\terr = timer(&bcr, e.PtransformID, e.TimerFamilyID)\n\t\t\t}\n\t\t\tif err != nil && err != io.EOF {\n\t\t\t\treturn errors.Wrap(err, \"source failed processing timers\")\n\t\t\t}\n\t\t\t// io.EOF means the reader successfully drained.\n\t\t\t// We're ready for a new buffer.\n\t\tcase <-ctx.Done():\n\t\t\t// now that it is done processing received data, we set it to false.\n\t\t\tn.consumingReceivedData.Store(false)\n\t\t\treturn nil\n\t\t}\n\t}\n}\n\n// ByteCountReader is a passthrough reader that counts all the bytes read through it.\n// It trusts the nested reader to return accurate byte information.\ntype byteCountReader struct {\n\tcount  *int\n\treader io.Reader\n}\n","sourceCodeStart":137,"sourceCodeEnd":173,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/datasource.go#L137-L173","documentation":"Same wrapping pattern as data errors but for timers: when a bundle event carries timers (len(e.Timers)>0), the `timer` callback processes them; any non-EOF error is wrapped as 'source failed processing timers'. It indicates the timer-processing path (DoFn ProcessElement with timer invocation, or timer decode) failed.","triggerScenarios":"An event with e.Timers set arrives; r.Reset(e.Timers) then timer(&bcr, e.PtransformID, e.TimerFamilyID) returns a non-nil error other than io.EOF during timer decoding or DoFn timer invocation.","commonSituations":"Stateful/timer DoFns (e.g. windowed aggregation with timers) failing at runtime; timer family mismatch after pipeline update; user code erroring inside timer-handling ProcessElement; runner re-delivering timers with stale schema.","solutions":["Read the wrapped cause error to identify the failing timer callback","Verify timer family IDs and coders match between pipeline stages / after job update","Fix the user DoFn logic executed when the timer fires","Check for incompatibilities if the pipeline was updated while draining state"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// timer callback errors surface as 'source failed processing timers'\nif err := timerFn(ctx, key); err != nil {\n    log.Errorf(\"timer processing failed: %v\", err)\n    return err\n}","preventionTips":["Test timer-heavy DoFns (stateful pipelines) before production","Keep timer family IDs stable across pipeline updates","Avoid state that can fail during timer firing; guard with nil checks","Watch for pipeline-update incompatibilities with existing timers"],"tags":["beam","go","timers"],"backgroundTag":"invalid-argument-value","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}