{"record":{"id":"71253c381d47540c","repo":"apache/beam","slug":"directrunner-does-not-support-duration-argument","errorCode":null,"errorMessage":"DirectRunner does not support duration argument.","messagePattern":"DirectRunner does not support duration argument\\.","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/direct/direct_runner.py","lineNumber":649,"sourceCode":"  def __init__(self, executor, evaluation_context):\n    super().__init__(PipelineState.RUNNING)\n    self._executor = executor\n    self._evaluation_context = evaluation_context\n\n  def __del__(self):\n    if self._state == PipelineState.RUNNING:\n      _LOGGER.warning(\n          'The DirectPipelineResult is being garbage-collected while the '\n          'DirectRunner is still running the corresponding pipeline. This may '\n          'lead to incomplete execution of the pipeline if the main thread '\n          'exits before pipeline completion. Consider using '\n          'result.wait_until_finish() to wait for completion of pipeline '\n          'execution.')\n\n  def wait_until_finish(self, duration=None):\n    if not PipelineState.is_terminal(self.state):\n      if duration:\n        raise NotImplementedError(\n            'DirectRunner does not support duration argument.')\n      try:\n        self._executor.await_completion()\n        self._state = PipelineState.DONE\n      except:  # pylint: disable=broad-except\n        self._state = PipelineState.FAILED\n        raise\n    return self._state\n\n  def aggregated_values(self, aggregator_or_name):\n    return self._evaluation_context.get_aggregator_values(aggregator_or_name)\n\n  def metrics(self):\n    return self._evaluation_context.metrics()\n\n  def cancel(self):\n    \"\"\"Shuts down pipeline workers.\n","sourceCodeStart":631,"sourceCodeEnd":667,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/direct/direct_runner.py#L631-L667","documentation":"DirectRunnerPipelineResult.wait_until_finish raises NotImplementedError if a duration argument is supplied, because the direct runner cannot impose a time limit on executor completion. Waiting without a duration is supported; bounded waiting is not.","triggerScenarios":"Calling result.wait_until_finish(duration=60) on a DirectRunner pipeline result while the pipeline is still running.","commonSituations":"Code shared across runners passing duration unconditionally; porting Dataflow-style timeout logic to local direct-runner tests.","solutions":["Call wait_until_finish() with no duration on the DirectRunner.","If a timeout is required, implement it in the caller (e.g. threading with join(timeout) or a watchdog) and cancel via result.cancel().","Only pass duration on runners that support it (e.g. Dataflow)."],"exampleFix":"// before\nresult.wait_until_finish(duration=120)  # DirectRunner -> NotImplementedError\n// after\nif runner == 'DirectRunner':\n  result.wait_until_finish()\nelse:\n  result.wait_until_finish(duration=120)","handlingStrategy":"try-catch","validationCode":"if isinstance(result, DirectRunnerPipelineResult) and duration is not None:\n    logging.warning('DirectRunner ignores duration; waiting indefinitely')\n    duration = None","typeGuard":"def supports_wait_duration(result) -> bool:\n    return type(result).__module__ != 'apache_beam.runners.direct.direct_runner'","tryCatchPattern":"try:\n    result.wait_until_finish(duration=duration)\nexcept NotImplementedError:\n    result.wait_until_finish()  # direct runner: no bounded wait","preventionTips":["Gate duration usage on the runner type.","Implement caller-side timeouts for local tests if needed.","Keep direct-runner and remote-runner wait logic in separate helpers."],"tags":["python","apache-beam","direct-runner","not-implemented"],"backgroundTag":"method-not-implemented","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"}