apache/beam · error · NotImplementedError

DirectRunner does not support duration argument.

Error message

DirectRunner does not support duration argument.

What it means

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.

Solutions

1. Call wait_until_finish() with no duration on the DirectRunner.
2. If a timeout is required, implement it in the caller (e.g. threading with join(timeout) or a watchdog) and cancel via result.cancel().
3. Only pass duration on runners that support it (e.g. Dataflow).

Example fix

// before
result.wait_until_finish(duration=120)  # DirectRunner -> NotImplementedError
// after
if runner == 'DirectRunner':
  result.wait_until_finish()
else:
  result.wait_until_finish(duration=120)

if isinstance(result, DirectRunnerPipelineResult) and duration is not None:
    logging.warning('DirectRunner ignores duration; waiting indefinitely')
    duration = None

def supports_wait_duration(result) -> bool:
    return type(result).__module__ != 'apache_beam.runners.direct.direct_runner'

try:
    result.wait_until_finish(duration=duration)
except NotImplementedError:
    result.wait_until_finish()  # direct runner: no bounded wait

When it happens

Trigger: Calling result.wait_until_finish(duration=60) on a DirectRunner pipeline result while the pipeline is still running.

Common situations: Code shared across runners passing duration unconditionally; porting Dataflow-style timeout logic to local direct-runner tests.

Related errors

• Assigning an index is not yet supported. Consider using…
• by
• collecting metrics will come later!
• concat(ignore_index)
• concat(levels)

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/71253c381d47540c. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/runners/direct/direct_runner.py:649

  def __init__(self, executor, evaluation_context):
    super().__init__(PipelineState.RUNNING)
    self._executor = executor
    self._evaluation_context = evaluation_context

  def __del__(self):
    if self._state == PipelineState.RUNNING:
      _LOGGER.warning(
          'The DirectPipelineResult is being garbage-collected while the '
          'DirectRunner is still running the corresponding pipeline. This may '
          'lead to incomplete execution of the pipeline if the main thread '
          'exits before pipeline completion. Consider using '
          'result.wait_until_finish() to wait for completion of pipeline '
          'execution.')

  def wait_until_finish(self, duration=None):
    if not PipelineState.is_terminal(self.state):
      if duration:
        raise NotImplementedError(
            'DirectRunner does not support duration argument.')
      try:
        self._executor.await_completion()
        self._state = PipelineState.DONE
      except:  # pylint: disable=broad-except
        self._state = PipelineState.FAILED
        raise
    return self._state

  def aggregated_values(self, aggregator_or_name):
    return self._evaluation_context.get_aggregator_values(aggregator_or_name)

  def metrics(self):
    return self._evaluation_context.metrics()

  def cancel(self):
    """Shuts down pipeline workers.

View on GitHub (pinned to 12126d8942)