apache/beam · error · RuntimeError

Trying to partition a cleared ListBuffer.

Error message

Trying to partition a cleared ListBuffer.

What it means

ListBuffer.partition() raises RuntimeError if the buffer was cleared. Partitioning a cleared buffer would operate on an empty/invalid _inputs list and break the buffer lifecycle contract, so it is rejected.

Solutions

1. Call reset() before partition() if reusing a cleared buffer.
2. Partition only buffers that are actively holding data for the current bundle.
3. Create a fresh ListBuffer instead of reusing cleared ones.

Example fix

# before
buf.clear()
parts = buf.partition(4)  # RuntimeError
# after
buf.clear()
buf.reset()
parts = buf.partition(4)

if buffer.cleared:
    buffer.reset()
parts = buffer.partition(n)

def is_partitionable(buf):
    return not buf.cleared

try:
    parts = buffer.partition(n)
except RuntimeError:
    buffer.reset()
    parts = buffer.partition(n)

When it happens

Trigger: Calling buffer.partition(n) after buffer.clear() — e.g. requesting partition slices of a reused buffer that was cleared but never reset().

Common situations: Runner-side code partitioning cached buffers between bundles without reset(); calling partition after teardown/cleanup already ran.

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 append to a cleared ListBuffer.
• Trying to iterate through 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/b280a01e99c8fe42. Report an issue: GitHub.

Appendix: source

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

  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)]
        idx = 0
        for input in self._inputs:
          input_stream = create_InputStream(input)
          while input_stream.size() > 0:
            decoded_value = self._coder_impl.decode_from_stream(
                input_stream, True)
            self._coder_impl.encode_to_stream(
                decoded_value, output_stream_list[idx], True)
            idx = (idx + 1) % n
        self._grouped_output = [[output_stream.get()]
                                for output_stream in output_stream_list]
      return self._grouped_output

View on GitHub (pinned to 12126d8942)