apache/beam · error · ValueError

Invalid state spec

Error message

Invalid state spec: %s

What it means

RuntimeStateFactory.for_spec maps userstate state specs (BagStateSpec, CombiningValueStateSpec, SetStateSpec, etc.) to direct-runner runtime state implementations. A state_spec of an unsupported type reaches the final else branch and raises ValueError. It usually means a new/unsupported state spec kind was used with the direct runner's streaming (TimelyQueue) execution.

Solutions

  1. Use a supported state spec: BagStateSpec, CombiningValueStateSpec, or SetStateSpec.
  2. Switch to a runner that supports the state spec you need (e.g. DataflowRunner or FlinkRunner).
  3. Upgrade apache-beam in case support for the spec was added in a newer release.

Example fix

// before
spec = ReadModifyWriteStateSpec('cache', coder)  # unmapped -> ValueError
// after
spec = userstate.BagStateSpec('cache', coder)  # or another supported spec
Defensive patterns

Strategy: validation

Validate before calling

SUPPORTED_SPECS = (userstate.BagStateSpec, userstate.CombiningValueStateSpec, userstate.SetStateSpec)
if not isinstance(state_spec, SUPPORTED_SPECS):
    raise ValueError(f'DirectRunner does not support state spec: {state_spec!r}')

Type guard

def is_supported_state_spec(spec) -> bool:
    import apache_beam.transforms.userstate as us
    return isinstance(spec, (us.BagStateSpec, us.CombiningValueStateSpec, us.SetStateSpec))

Try / catch

try:
    run_streaming_pipeline()
except ValueError as e:
    if 'Invalid state spec' in str(e):
        sys.exit('Restrict DoFn state to bag/combining/set specs on DirectRunner')
    raise

Prevention

When it happens

Trigger: Using DoFn state (e.g. ReadModifyWriteStateSpec, OrderedListStateSpec, or a custom UserStateSpec subclass) in a streaming DirectRunner pipeline where for_spec has no mapping for that spec type.

Common situations: Adopting a newer state API before the direct runner implements it; custom state spec implementations; running a pipeline locally that was built for a different runner.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/direct/direct_userstate.py:49

  def __init__(self, state_spec, state_tag, current_value_accessor):
    self._state_spec = state_spec
    self._state_tag = state_tag
    self._current_value_accessor = current_value_accessor

  @staticmethod
  def for_spec(state_spec, state_tag, current_value_accessor):
    if isinstance(state_spec, userstate.ReadModifyWriteStateSpec):
      return ReadModifyWriteRuntimeState(
          state_spec, state_tag, current_value_accessor)
    elif isinstance(state_spec, userstate.BagStateSpec):
      return BagRuntimeState(state_spec, state_tag, current_value_accessor)
    elif isinstance(state_spec, userstate.CombiningValueStateSpec):
      return CombiningValueRuntimeState(
          state_spec, state_tag, current_value_accessor)
    elif isinstance(state_spec, userstate.SetStateSpec):
      return SetRuntimeState(state_spec, state_tag, current_value_accessor)
    else:
      raise ValueError('Invalid state spec: %s' % state_spec)

  def _encode(self, value):
    return self._state_spec.coder.encode(value)

  def _decode(self, value):
    return self._state_spec.coder.decode(value)


# Sentinel designating an unread value.
UNREAD_VALUE = object()


class ReadModifyWriteRuntimeState(DirectRuntimeState,
                                  userstate.ReadModifyWriteRuntimeState):
  def __init__(self, state_spec, state_tag, current_value_accessor):
    super().__init__(state_spec, state_tag, current_value_accessor)
    self._value = UNREAD_VALUE
    self._cleared = False

View on GitHub (pinned to 12126d8942)