{"record":{"id":"4dc2317a7d31359b","repo":"apache/beam","slug":"processing-of-an-element-in-transform-v-has-exceeded-the","errorCode":null,"errorMessage":"Processing of an element in transform %v has exceeded the specified timeout of %v without outputting or completing in state %v, SDK harness will be terminated","messagePattern":"Processing of an element in transform (.+?) has exceeded the specified timeout of (.+?) without outputting or completing in state (.+?), SDK harness will be terminated","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/metrics/sampler.go","lineNumber":70,"sourceCode":"\tif v, ok := s.store.stateRegistry[ps.pid]; ok {\n\t\tv[ps.state].TotalTime += t\n\t\tv[TotalBundle].TotalTime += t\n\n\t\tif s.transitionsAtLastSample != ps.transitions {\n\t\t\t// state change detected\n\t\t\ts.millisSinceLastTransition = 0\n\t\t\ts.transitionsAtLastSample = ps.transitions\n\t\t\ts.nextLogTime = s.logInterval\n\t\t} else {\n\t\t\ts.millisSinceLastTransition += t\n\t\t}\n\n\t\tif s.millisSinceLastTransition > s.nextLogTime {\n\t\t\tlog.Infof(ctx, \"Operation ongoing in transform %v for at least %v without outputting or completing in state %v\", ps.pid, s.millisSinceLastTransition, getState(ps.state))\n\t\t\ts.nextLogTime += s.logInterval\n\t\t}\n\t\tif s.restartLullTimeout > 0 && s.millisSinceLastTransition > s.restartLullTimeout {\n\t\t\treturn errors.Errorf(\"Processing of an element in transform %v has exceeded the specified timeout of %v without outputting or completing in state %v, SDK harness will be terminated\", ps.pid, s.restartLullTimeout, getState(ps.state))\n\t\t}\n\t}\n\treturn nil\n}\n\n// SetLogInterval sets the logging interval for lull reporting.\nfunc (s *StateSampler) SetLogInterval(t time.Duration) {\n\ts.logInterval = t\n}\n\nfunc loadCurrentState(s *StateSampler) currentStateVal {\n\tts := (atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&s.store.bundleState))))\n\tif ts == nil {\n\t\treturn currentStateVal{}\n\t}\n\tbs := *(*BundleState)(ts)\n\treturn currentStateVal{pid: bs.pid, state: bs.currentState, transitions: atomic.LoadInt64(s.store.transitions)}\n}","sourceCodeStart":52,"sourceCodeEnd":88,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/metrics/sampler.go#L52-L88","documentation":"The Beam Go SDK's sampler periodically inspects in-flight bundle processing. When an element in a PTransform has been processing longer than the configured restart lull timeout (restartLullTimeout) with no output or state transition, the sampler returns this error to terminate the SDK harness, since the pipeline is presumed hung. It is a deliberate liveness watchdog, not a data-dependent bug.","triggerScenarios":"sampler.Sample is called while a transform's state has not transitioned for longer than s.restartLullTimeout; configured via a nonzero restartLullTimeout on the sampler (set when the harness enables lull detection).","commonSituations":"User DoFns blocked on slow I/O, deadlocks waiting on side inputs, extremely large single elements, or resource starvation (CPU/threads) in the SDK worker container causing processing to stall.","solutions":["Fix the underlying stall: profile the DoFn for blocking calls (unbounded waits, missing side input, deadlocked mutexes).","Increase or disable the restart lull timeout via the runner/harness lull-detection options (e.g. set it to 0 to disable) if long processing is legitimate.","Break large elements into smaller ones or add progress logging so the sampler sees state transitions.","Check worker logs around the reported pid/state to identify which transform and state stalled."],"exampleFix":"// before: DoFn blocks forever waiting for external service\nresp := <-ch // no timeout, hangs the bundle\n// after\nselect {\ncase resp := <-ch:\n    _ = resp\ncase <-time.After(30 * time.Second):\n    return errors.New(\"upstream call timed out\")\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// runner-side: handle harness termination and retry the bundle\nif err := bundle.Execute(ctx); err != nil {\n    if strings.Contains(err.Error(), \"exceeded the specified timeout\") {\n        log.Errorf(\"lull timeout in %s: %v\", transformID, err)\n        // alert / restart worker / adjust lull timeout config\n    }\n    return err\n}","preventionTips":["Never block indefinitely in DoFns; always use timeouts on I/O and channel waits.","Size workers adequately so elements don't stall from CPU/memory starvation.","Set a restart lull timeout appropriate for your slowest legitimate element.","Monitor sampler logs ('Operation ongoing in transform ...') as early warnings before the timeout fires."],"tags":["go","apache-beam","timeout","harness","lull-detection"],"backgroundTag":"request-timeout","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"}