apache/beam · error · RuntimeError

ListBuffer append after read.

Error message

ListBuffer append after read.

What it means

ListBuffer.extend() also refuses appends after the buffer's grouped output has been produced (self._grouped_output is set). Once results have been grouped/read, the buffer is considered finalized and further mutation would desynchronize the cached grouped output from _inputs, so a RuntimeError is raised.

Solutions

1. Finish all appends before calling partition()/reading grouped output.
2. Create a new ListBuffer for post-partition data instead of reusing the finalized one.
3. Call clear() then reset() to start over if the buffer must be reused.

Example fix

# before
out = buf.partition(4)
buf.extend(extra)  # RuntimeError
# after
buf.extend(extra)
out = buf.partition(4)

if buffer._grouped_output is None and not buffer.cleared:
    buffer.extend(other)

def is_extendable(buf):
    return not buf.cleared and not buf._grouped_output

try:
    buffer.extend(other)
except RuntimeError:
    new_buf = ListBuffer(buffer._coder_impl)
    new_buf.extend(other)

When it happens

Trigger: Calling extend() after partition(n) has been called and its grouped output was generated (or otherwise after _grouped_output was set), i.e. appending to an already-read buffer.

Common situations: Runner code that keeps appending elements after distributing/partitioning the buffer's contents; reusing a partitioned buffer for the next bundle without clearing/resetting.

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

• Grouping table append after read.
• Trying to append to a cleared ListBuffer.
• Trying to iterate through a cleared ListBuffer.
• Trying to partition a cleared ListBuffer.
• Trying to reset a non-cleared ListBuffer.

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py:136

class ListBuffer:
  """Used to support parititioning of a list."""
  def __init__(self, coder_impl: Optional[CoderImpl]) -> None:
    self._coder_impl = coder_impl or CoderImpl()
    self._inputs: list[bytes] = []
    self._grouped_output: Optional[list[list[bytes]]] = None
    self.cleared = False

  def copy(self) -> 'ListBuffer':
    new = ListBuffer(self._coder_impl)
    new._inputs = [v for v in self._inputs]
    return new

  def extend(self, extra: 'Buffer') -> None:
    if self.cleared:
      raise RuntimeError('Trying to append to a cleared ListBuffer.')
    if self._grouped_output:
      raise RuntimeError('ListBuffer append after read.')
    assert isinstance(extra, ListBuffer)
    self._inputs.extend(extra._inputs)

  def append(self, element: bytes) -> None:
    if self.cleared:
      raise RuntimeError('Trying to append to a cleared ListBuffer.')
    if self._grouped_output:
      raise RuntimeError('ListBuffer append after read.')
    self._inputs.append(element)

  def partition(self, n: int) -> list[list[bytes]]:
    if self.cleared:
      raise RuntimeError('Trying to partition a cleared ListBuffer.')
    if len(self._inputs) >= n or len(self._inputs) == 0:
      return [self._inputs[k::n] for k in range(n)]
    else:
      if not self._grouped_output:
        output_stream_list = [create_OutputStream() for _ in range(n)]

View on GitHub (pinned to 12126d8942)