{"record":{"id":"f0f02ae22c01c93e","repo":"apache/beam","slug":"input-is-already-set","errorCode":null,"errorMessage":"input is already set.","messagePattern":"input is already set\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/portability/fn_api_runner/worker_handlers.py","lineNumber":159,"sourceCode":"      return None\n    if not req.instruction_id:\n      with ControlConnection._lock:\n        ControlConnection._uid_counter += 1\n        req.instruction_id = 'control_%s' % ControlConnection._uid_counter\n    future = ControlFuture(req.instruction_id)\n    self._futures_by_id[req.instruction_id] = future\n    self._push_queue.put(req)\n    return future\n\n  def get_req(self):\n    # type: () -> Union[Sentinel, beam_fn_api_pb2.InstructionRequest]\n    return self._push_queue.get()\n\n  def set_input(self, input):\n    # type: (Iterable[beam_fn_api_pb2.InstructionResponse]) -> None\n    with ControlConnection._lock:\n      if self._input:\n        raise RuntimeError('input is already set.')\n      self._input = input\n      self._read_thread.start()\n      self._state = BeamFnControlServicer.STARTED_STATE\n\n  def close(self):\n    # type: () -> None\n    with ControlConnection._lock:\n      if self._state == BeamFnControlServicer.STARTED_STATE:\n        self.push(BeamFnControlServicer._DONE_MARKER)\n        self._read_thread.join()\n      self._state = BeamFnControlServicer.DONE_STATE\n\n  def abort(self, exn):\n    # type: (Exception) -> None\n    for future in self._futures_by_id.values():\n      future.abort(exn)\n\n","sourceCodeStart":141,"sourceCodeEnd":177,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/portability/fn_api_runner/worker_handlers.py#L141-L177","documentation":"`ControlConnection.set_input` raises RuntimeError('input is already set.') when the connection's input iterable is assigned twice. The input queue is a one-time wiring: once set, the read thread starts and the connection transitions to STARTED_STATE, so rebinding is prohibited under `ControlConnection._lock`.","triggerScenarios":"Calling `control_connection.set_input(...)` a second time on the same connection, or calling it after the connection was already started/used (e.g. reusing a cached connection across worker runs).","commonSituations":"Worker handler reuse in the Fn API runner; retry logic that re-invokes setup on the same ControlConnection instead of creating a new one; race where two threads initialize the same connection.","solutions":["Call set_input exactly once per ControlConnection; create a new connection for a new input stream","Guard the call with a check of the connection state before re-wiring","Fix lifecycle code that caches and reuses connections across worker restarts","Serialize initialization (e.g. only the controlling thread calls set_input) to avoid races"],"exampleFix":"// before\nconn.set_input(new_input)  # RuntimeError if already set\n// after\nif not conn._input:\n  conn.set_input(new_input)\nelse:\n  conn = ControlConnection(...)\n  conn.set_input(new_input)","handlingStrategy":"try-catch","validationCode":"if getattr(conn, '_input', None):\n  raise RuntimeError('set_input already called on this connection')","typeGuard":"def can_set_input(conn):\n  return not conn._input and conn._state != BeamFnControlServicer.STARTED_STATE","tryCatchPattern":"try:\n  conn.set_input(input)\nexcept RuntimeError as e:\n  if 'input is already set' in str(e):\n    conn = ControlConnection(...)  # fresh connection\n    conn.set_input(input)\n  else:\n    raise","preventionTips":["Call set_input once per ControlConnection lifetime","Guard with a lock/flag if multiple threads may initialize","Do not cache/reuse connections across worker restarts"],"tags":["python","apache-beam","grpc","control-connection","state"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}