{"record":{"id":"8eb9aaf3844b6930","repo":"apache/beam","slug":"causedbydrain-unsupported-in-s","errorCode":null,"errorMessage":"CausedByDrain unsupported in %s","messagePattern":"CausedByDrain unsupported in (.+?)","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnInvoker.java","lineNumber":372,"sourceCode":"      throw new UnsupportedOperationException(\n          String.format(\"RecordOffset unsupported in %s\", getErrorContext()));\n    }\n\n    @Override\n    public Instant fireTimestamp(DoFn<InputT, OutputT> doFn) {\n      throw new UnsupportedOperationException(\n          String.format(\"FireTimestamp unsupported in %s\", getErrorContext()));\n    }\n\n    @Override\n    public CausedByDrain causedByDrain(DoFn<InputT, OutputT> doFn) {\n      throw new UnsupportedOperationException(\n          String.format(\"CausedByDrain unsupported in %s\", getErrorContext()));\n    }\n\n    @Override\n    public ValueKind valueKind(DoFn<InputT, OutputT> doFn) {\n      throw new UnsupportedOperationException(\n          String.format(\"ValueKind unsupported in %s\", getErrorContext()));\n    }\n\n    @Override\n    public String timerId(DoFn<InputT, OutputT> doFn) {\n      throw new UnsupportedOperationException(\n          String.format(\"TimerId unsupported in %s\", getErrorContext()));\n    }\n\n    @Override\n    public TimeDomain timeDomain(DoFn<InputT, OutputT> doFn) {\n      throw new UnsupportedOperationException(\n          String.format(\"TimeDomain unsupported in %s\", getErrorContext()));\n    }\n\n    @Override\n    public OutputReceiver<OutputT> outputReceiver(DoFn<InputT, OutputT> doFn) {\n      throw new UnsupportedOperationException(","sourceCodeStart":354,"sourceCodeEnd":390,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnInvoker.java#L354-L390","documentation":"This error is thrown by the default UnsupportedInvocationBehavior in DoFnInvoker when causedByDrain() is called from a context that cannot report drain causality. causedByDrain() tells an @OnTimer callback whether the timer fired because of pipeline draining (Drain job action) and is only meaningful on runners that implement drain semantics. Outside such a runner — or when the invoker lacks drain support — the call is rejected rather than returning a misleading false.","triggerScenarios":"Invoking context.causedByDrain() from an @OnTimer callback while the DoFn runs under an invoker with UnsupportedInvocationBehavior (direct DoFnInvokers.invokerFor(...) usage, unit-test harness, or a runner without drain support).","commonSituations":"Unit-testing a timer callback that checks causedByDrain(); running on a runner or in a Batch mode where draining is not implemented; invoking the DoFn through a custom/legacy DoFnRunner that doesn't wire up the CausedByDrain value.","solutions":["Only call causedByDrain() from @OnTimer callbacks and only when targeting runners that support drain (e.g. Dataflow Streaming)","Remove the check if drain semantics are irrelevant to your pipeline","In tests, avoid asserting drain behavior via direct invoker usage; test drain paths with the actual runner or mock the context interface explicitly","Wrap the call in try-catch for UnsupportedOperationException and treat it as not-caused-by-drain when unsupported"],"exampleFix":"// before\nif (ctx.causedByDrain()) { flush(); }\n// after\nboolean drain;\ntry { drain = ctx.causedByDrain(); } catch (UnsupportedOperationException e) { drain = false; }\nif (drain) { flush(); }","handlingStrategy":"fallback","validationCode":"// Only query drain causality in streaming @OnTimer callbacks on runners with drain support;\n// otherwise skip the check entirely.","typeGuard":"boolean safeCausedByDrain(DoFn.OnTimerContext ctx) {\n  try { return ctx.causedByDrain(); }\n  catch (UnsupportedOperationException e) { return false; }\n}","tryCatchPattern":"try {\n  if (context.causedByDrain()) { flushAndAcknowledgeDrain(); }\n} catch (UnsupportedOperationException e) {\n  // runner does not support drain; continue normal processing\n}","preventionTips":["Gate drain-handling code behind runner capability (e.g. only Dataflow Streaming drains jobs)","Never call causedByDrain() from @ProcessElement — it is timer-context only","Don't rely on drain detection in unit tests; test drain behavior on the actual runner"],"tags":["java","apache-beam","dofn","timers","drain","unsupported-operation"],"backgroundTag":"unsupported-operation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}