What causes "{timer.time_domain}" in apache/beam?

Change the timer to EVENT_TIME or PROCESSING_TIME domain

apache/beam · error · NotImplementedError

{timer.time_domain}

Error message

{timer.time_domain}

What it means

check_requirements() inspects ParDo payloads for timer family specs and rejects timers declared in a time domain other than EVENT_TIME or PROCESSING_TIME. The unsupported time_domain enum value is raised as NotImplementedError.

Solutions

Change the timer to EVENT_TIME or PROCESSING_TIME domain
Upgrade to a runner version that supports the timer time domain you need
Drop or refactor the timer usage if the domain isn't required

Example fix

// before
@user_timer(time_domain=TimeDomain.SYNCHRONIZED_PROCESSING_TIME)
def my_timer(self):
  ...
// after
@user_timer(time_domain=TimeDomain.PROCESSING_TIME)
def my_timer(self):
  ...

Defensive patterns

Strategy: validation

Validate before calling

from apache_beam.portability.api import beam_runner_api_pb2
allowed = (beam_runner_api_pb2.TimeDomain.EVENT_TIME,
           beam_runner_api_pb2.TimeDomain.PROCESSING_TIME)
assert all(t.time_domain in allowed
           for t in timer_family_specs.values())

Try / catch

try:
    runner.check_requirements(pipeline_proto, runner.supported_requirements)
except NotImplementedError as e:
    print('unsupported timer time domain:', e)

Prevention

Declare timers only with EVENT_TIME or PROCESSING_TIME unless the runner supports more
Review timer time domains when porting between runners
Test timer-bearing pipelines on the target runner early

When it happens

Trigger: A DoFn declares a @user_timer (or timer family) with a time_domain outside {EVENT_TIME, PROCESSING_TIME} — e.g. a runner-unsupported or SYNCHRONIZED_PROCESSING_TIME domain — and the pipeline is run through run_portable_pipeline.

Common situations: Using synchronized processing-time timers in a runner that hasn't implemented them; porting a pipeline from a runner supporting all time domains to one that supports only event/processing time.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/runner.py:226

    # pylint: disable=wrong-import-order, wrong-import-position
    from apache_beam.runners.portability.fn_api_runner import translations
    supported_requirements = set(supported_requirements)
    for requirement in pipeline_proto.requirements:
      if requirement not in supported_requirements:
        raise ValueError(
            'Unable to run pipeline with requirement: %s' % requirement)
    for transform in pipeline_proto.components.transforms.values():
      if transform.spec.urn == common_urns.primitives.TEST_STREAM.urn:
        if common_urns.primitives.TEST_STREAM.urn not in supported_requirements:
          raise NotImplementedError(transform.spec.urn)
      elif transform.spec.urn in translations.PAR_DO_URNS:
        payload = beam_runner_api_pb2.ParDoPayload.FromString(
            transform.spec.payload)
        for timer in payload.timer_family_specs.values():
          if timer.time_domain not in (
              beam_runner_api_pb2.TimeDomain.EVENT_TIME,
              beam_runner_api_pb2.TimeDomain.PROCESSING_TIME):
            raise NotImplementedError(timer.time_domain)

  def default_pickle_library_override(self):
    """Default pickle library, can be overridden by runner implementation."""
    return None


# FIXME: replace with PipelineState(str, enum.Enum)
class PipelineState(object):
  """State of the Pipeline, as returned by :attr:`PipelineResult.state`.

  This is meant to be the union of all the states any runner can put a
  pipeline in. Currently, it represents the values of the dataflow
  API JobState enum.
  """
  UNKNOWN = 'UNKNOWN'  # not specified by a runner, or unknown to a runner.
  STARTING = 'STARTING'  # not yet started
  STOPPED = 'STOPPED'  # paused or not yet started
  RUNNING = 'RUNNING'  # currently running

View on GitHub (pinned to 12126d8942)