apache/beam · error · RuntimeError

input is already set.

Error message

input is already set.

What it means

`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`.

Source

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

      return None
    if not req.instruction_id:
      with ControlConnection._lock:
        ControlConnection._uid_counter += 1
        req.instruction_id = 'control_%s' % ControlConnection._uid_counter
    future = ControlFuture(req.instruction_id)
    self._futures_by_id[req.instruction_id] = future
    self._push_queue.put(req)
    return future

  def get_req(self):
    # type: () -> Union[Sentinel, beam_fn_api_pb2.InstructionRequest]
    return self._push_queue.get()

  def set_input(self, input):
    # type: (Iterable[beam_fn_api_pb2.InstructionResponse]) -> None
    with ControlConnection._lock:
      if self._input:
        raise RuntimeError('input is already set.')
      self._input = input
      self._read_thread.start()
      self._state = BeamFnControlServicer.STARTED_STATE

  def close(self):
    # type: () -> None
    with ControlConnection._lock:
      if self._state == BeamFnControlServicer.STARTED_STATE:
        self.push(BeamFnControlServicer._DONE_MARKER)
        self._read_thread.join()
      self._state = BeamFnControlServicer.DONE_STATE

  def abort(self, exn):
    # type: (Exception) -> None
    for future in self._futures_by_id.values():
      future.abort(exn)

View on GitHub (pinned to 12126d8942)

Solutions

1. Call set_input exactly once per ControlConnection; create a new connection for a new input stream
2. Guard the call with a check of the connection state before re-wiring
3. Fix lifecycle code that caches and reuses connections across worker restarts
4. Serialize initialization (e.g. only the controlling thread calls set_input) to avoid races

Example fix

// before
conn.set_input(new_input)  # RuntimeError if already set
// after
if not conn._input:
  conn.set_input(new_input)
else:
  conn = ControlConnection(...)
  conn.set_input(new_input)

if getattr(conn, '_input', None):
  raise RuntimeError('set_input already called on this connection')

def can_set_input(conn):
  return not conn._input and conn._state != BeamFnControlServicer.STARTED_STATE

try:
  conn.set_input(input)
except RuntimeError as e:
  if 'input is already set' in str(e):
    conn = ControlConnection(...)  # fresh connection
    conn.set_input(input)
  else:
    raise

When it happens

Trigger: 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).

Common situations: 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.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• {module} is not registered for pickle by value
• PCollection not available, please run the pipeline.
• Buffer size {size} exceeds GRPC limit {_FLUSH_MAX_SIZE}. Thi
• response.error
• Header metadata already has a worker_id.

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