apache/beam · error · NotImplementedError

Unknown state type:

Error message

Unknown state type: 

What it means

The StateServicer handling BeamFnState RPCs only supports a fixed set of state_key oneof types (its _SUPPORTED_STATE_TYPES). If a request carries a state key type outside that set, it raises NotImplementedError naming the unknown type. This guards against newer/other state kinds the local servicer cannot serve.

Solutions

1. Use a state API type supported by the runner (e.g. user BagState/timer types it supports).
2. Upgrade the runner (or whole Beam) so SDK and runner versions match and support the state type.
3. Switch to a runner that supports the needed state kind (e.g. Dataflow/Flink) if required.
4. Check which oneof type is named in the message and consult _SUPPORTED_STATE_TYPES.

Example fix

// before
state = ReadModifyWriteStateSpec(...)
// after (if runner only supports bag state)
state = BagStateSpec('state', coders.StrUtf8Coder())

if state_key.WhichOneof('type') not in StateServicer._SUPPORTED_STATE_TYPES:
    raise UnsupportedStateType(state_key.WhichOneof('type'))

def supported_state(key, supported): return key.WhichOneof('type') in supported

try:
    resp = stub.State(iter_requests)
except Exception as e:
    if 'Unknown state type' in str(e): fall_back_to_supported_state_api()

When it happens

Trigger: A pipeline using user-state features (e.g. BagState, MultimapState, OrderedListState or a newer state type) whose StateKey oneof is not in _SUPPORTED_STATE_TYPES is executed against a runner using this servicer, via GetState/Append state requests.

Common situations: Using stateful DoFn APIs unsupported by the Fn API runner/portable runner combination; SDK newer than runner introducing new state types; cross-version SDK/runner mismatch.

Related errors

• Unknown state request
• input is already set.
• response.error
• State stream is closed.
• This SDK is only capable of dealing with

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/portability/fn_api_runner/worker_handlers.py:1089

    yield

  def _get_one_interval_key(self, state_key, start):
    # type: (beam_fn_api_pb2.StateKey, int) -> bytes
    state_key_copy = beam_fn_api_pb2.StateKey()
    state_key_copy.CopyFrom(state_key)
    state_key_copy.ordered_list_user_state.range.start = start
    state_key_copy.ordered_list_user_state.range.end = start + 1
    return self._to_key(state_key_copy)

  def get_raw(
      self,
      state_key,  # type: beam_fn_api_pb2.StateKey
      continuation_token=None  # type: Optional[bytes]
  ):
    # type: (...) -> Tuple[bytes, Optional[bytes]]

    if state_key.WhichOneof('type') not in self._SUPPORTED_STATE_TYPES:
      raise NotImplementedError(
          'Unknown state type: ' + state_key.WhichOneof('type'))  # type: ignore[operator]

    with self._lock:
      if not continuation_token:
        # Compute full_state only when no continuation token is provided.
        # If there is continuation token, full_state is already in
        # continuation cache. No need to recompute.
        full_state = []  # type: List[bytes]
        if state_key.WhichOneof('type') == 'ordered_list_user_state':
          maybe_start = state_key.ordered_list_user_state.range.start
          maybe_end = state_key.ordered_list_user_state.range.end
          persistent_state_key = beam_fn_api_pb2.StateKey()
          persistent_state_key.CopyFrom(state_key)
          persistent_state_key.ordered_list_user_state.ClearField("range")

          available_keys = self._ordered_list_keys[self._to_key(
              persistent_state_key)]

View on GitHub (pinned to 12126d8942)