apache/beam · error · NotImplementedError

Unknown state request

Error message

Unknown state request: %s

What it means

The StateServicer's request loop dispatches on request.WhichOneof('request') values (get/append/clear). Any other request type reaches the else branch and raises NotImplementedError. Notably the surrounding try/except converts this into a StateResponse with an error field rather than crashing the stream.

Solutions

1. Align SDK harness and runner versions so both speak the same BeamFnApi proto revision.
2. Regenerate/use matching beam_fn_api_pb2 on both sides.
3. Inspect the StateResponse error payload / logged request_type to identify the unsupported request kind.
4. If implementing a custom handler, add the request type to the dispatch chain.

rt = request.WhichOneof('request')
if rt not in ('get', 'append', 'clear'):
    skip_request(rt)

def known_state_request(req): return req.WhichOneof('request') in {'get','append','clear'}

for resp in stub.State(iter_requests):
    if resp.error:
        handle_state_error(resp.error)

When it happens

Trigger: An SDK harness sends a BeamFnStateRequest whose request oneof is a type this servicer doesn't handle (newer state request kind, or corrupted request) over the state gRPC stream.

Common situations: Version mismatch where a newer SDK harness issues state requests an older local job server doesn't implement; custom/generated protobuf stubs out of sync with the runner's proto.

Related errors

• Unknown state type:
• 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/a74338d9c85bc6f9. Report an issue: GitHub.

Appendix: source

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

      try:
        request_type = request.WhichOneof('request')
        if request_type == 'get':
          data, continuation_token = self._state.get_raw(
              request.state_key, request.get.continuation_token)
          yield beam_fn_api_pb2.StateResponse(
              id=request.id,
              get=beam_fn_api_pb2.StateGetResponse(
                  data=data, continuation_token=continuation_token))
        elif request_type == 'append':
          self._state.append_raw(request.state_key, request.append.data)
          yield beam_fn_api_pb2.StateResponse(
              id=request.id, append=beam_fn_api_pb2.StateAppendResponse())
        elif request_type == 'clear':
          self._state.clear(request.state_key)
          yield beam_fn_api_pb2.StateResponse(
              id=request.id, clear=beam_fn_api_pb2.StateClearResponse())
        else:
          raise NotImplementedError('Unknown state request: %s' % request_type)
      except Exception as exn:
        yield beam_fn_api_pb2.StateResponse(id=request.id, error=str(exn))


class SingletonStateHandlerFactory(sdk_worker.StateHandlerFactory):
  """A singleton cache for a StateServicer."""
  def __init__(self, state_handler):
    # type: (sdk_worker.CachingStateHandler) -> None
    self._state_handler = state_handler

  def create_state_handler(self, api_service_descriptor):
    # type: (endpoints_pb2.ApiServiceDescriptor) -> sdk_worker.CachingStateHandler

    """Returns the singleton state handler."""
    return self._state_handler

  def close(self):
    # type: () -> None

View on GitHub (pinned to 12126d8942)