apache/beam · error · RuntimeError

Trying to append to a cleared ListBuffer.

Error message

Trying to append to a cleared ListBuffer.

What it means

ListBuffer.extend() refuses to append another Buffer's inputs once this buffer has been cleared via clear(). After clear(), _inputs is emptied and cleared=True marks the buffer dead; extending it would corrupt the bundle-buffer reuse protocol used by the fn_api_runner. A RuntimeError is raised to surface the lifecycle violation.

Solutions

1. Call reset() on the cleared buffer before extending it again, or create a fresh ListBuffer.
2. Restructure code so clear() is the last operation on a buffer (clear only when done).
3. Track buffer lifecycle explicitly (e.g. set reference to None after clear) to avoid accidental reuse.

Example fix

# before
buf.clear()
buf.extend(other)  # RuntimeError
# after
buf.clear()
buf.reset()
buf.extend(other)

if buffer.cleared:
    buffer.reset()
buffer.extend(other)

def is_writable(buf):
    return not buf.cleared and not getattr(buf, '_grouped_output', None)

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

When it happens

Trigger: Calling buffer.extend(other) after buffer.clear() has been called (cleared is True), typically when reusing a cached Buffer object across bundle executions instead of calling reset() first.

Common situations: Runner-side code caching ListBuffer instances between bundles and forgetting reset(); custom runner extensions that clear a buffer then continue writing; race between a clearing thread and an appending thread.

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.
• ListBuffer append after read.
• 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/48e874a9c78d5940. Report an issue: GitHub.

Appendix: source

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

    pass


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:

View on GitHub (pinned to 12126d8942)