{"record":{"id":"06594b86040fc58c","repo":"apache/beam","slug":"source-failed-processing-data","errorCode":null,"errorMessage":"source failed processing data","messagePattern":"source failed processing data","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/datasource.go","lineNumber":146,"sourceCode":"\tvar byteCount int\n\tbcr := byteCountReader{reader: &r, count: &byteCount}\n\n\tfor {\n\t\tn.consumingReceivedData.Store(false)\n\t\tvar err error\n\t\tselect {\n\t\tcase e, ok := <-elms:\n\t\t\tn.consumingReceivedData.Store(true)\n\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}","sourceCodeStart":128,"sourceCodeEnd":164,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/datasource.go#L128-L164","documentation":"The DataSource process() loop wraps any error returned by the user DoFn's process callback while consuming a data element. The underlying error is preserved via errors.Wrapf; io.EOF is excluded because it signals a successfully drained reader, not a failure. This means the actual failure is a user DoFn error or a nested decode/process error, surfaced at the source boundary.","triggerScenarios":"A data element arrives on the runner's data channel with len(e.Data)>0; r.Reset(e.Data) succeeds but the `data(&bcr, e.PtransformID)` callback (which decodes elements and invokes the DoFn) returns a non-nil error other than io.EOF.","commonSituations":"User DoFn panics/returns errors inside processElement; custom coders failing to decode; downstream transform raising errors on specific records; corrupted element payloads from an incompatible pipeline version.","solutions":["Inspect the wrapped (cause) error in the message chain to find the failing DoFn or decoder","Fix the user DoFn code that returned the error for the offending record","Re-verify coder consistency between pipeline construction and runtime (same SDK versions)","If caused by a specific element, filter or handle bad records defensively in the DoFn"],"exampleFix":"// before\nfunc (fn *myFn) ProcessElement(v string) error {\n    return doSomething(v) // may fail on bad input\n}\n// after\nfunc (fn *myFn) ProcessElement(v string) error {\n    if err := doSomething(v); err != nil {\n        // handle or fall back instead of failing the bundle\n        return handleGracefully(v, err)\n    }\n    return nil\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// DoFn process errors propagate as wrapped 'source failed processing data'\nif err := dofn.ProcessElement(v); err != nil {\n    // log and decide: fail bundle vs. skip record\n    log.Errorf(\"element %v failed: %v\", v, err)\n    return err // or return nil to skip\n}","preventionTips":["Make DoFns defensive against malformed/unexpected input values","Return nil for skip-worthy bad records instead of failing the whole bundle","Keep coder definitions in sync across pipeline stages","Add unit tests for DoFn edge cases"],"tags":["beam","go","data-processing"],"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"}