{"record":{"id":"6f674274efd789d7","repo":"apache/beam","slug":"listbuffer-append-after-read","errorCode":null,"errorMessage":"ListBuffer append after read.","messagePattern":"ListBuffer append after read\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py","lineNumber":136,"sourceCode":"\nclass ListBuffer:\n  \"\"\"Used to support parititioning of a list.\"\"\"\n  def __init__(self, coder_impl: Optional[CoderImpl]) -> None:\n    self._coder_impl = coder_impl or CoderImpl()\n    self._inputs: list[bytes] = []\n    self._grouped_output: Optional[list[list[bytes]]] = None\n    self.cleared = False\n\n  def copy(self) -> 'ListBuffer':\n    new = ListBuffer(self._coder_impl)\n    new._inputs = [v for v in self._inputs]\n    return new\n\n  def extend(self, extra: 'Buffer') -> None:\n    if self.cleared:\n      raise RuntimeError('Trying to append to a cleared ListBuffer.')\n    if self._grouped_output:\n      raise RuntimeError('ListBuffer append after read.')\n    assert isinstance(extra, ListBuffer)\n    self._inputs.extend(extra._inputs)\n\n  def append(self, element: bytes) -> None:\n    if self.cleared:\n      raise RuntimeError('Trying to append to a cleared ListBuffer.')\n    if self._grouped_output:\n      raise RuntimeError('ListBuffer append after read.')\n    self._inputs.append(element)\n\n  def partition(self, n: int) -> list[list[bytes]]:\n    if self.cleared:\n      raise RuntimeError('Trying to partition a cleared ListBuffer.')\n    if len(self._inputs) >= n or len(self._inputs) == 0:\n      return [self._inputs[k::n] for k in range(n)]\n    else:\n      if not self._grouped_output:\n        output_stream_list = [create_OutputStream() for _ in range(n)]","sourceCodeStart":118,"sourceCodeEnd":154,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py#L118-L154","documentation":"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.","triggerScenarios":"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.","commonSituations":"Runner code that keeps appending elements after distributing/partitioning the buffer's contents; reusing a partitioned buffer for the next bundle without clearing/resetting.","solutions":["Finish all appends before calling partition()/reading grouped output.","Create a new ListBuffer for post-partition data instead of reusing the finalized one.","Call clear() then reset() to start over if the buffer must be reused."],"exampleFix":"# before\nout = buf.partition(4)\nbuf.extend(extra)  # RuntimeError\n# after\nbuf.extend(extra)\nout = buf.partition(4)","handlingStrategy":"type-guard","validationCode":"if buffer._grouped_output is None and not buffer.cleared:\n    buffer.extend(other)","typeGuard":"def is_extendable(buf):\n    return not buf.cleared and not buf._grouped_output","tryCatchPattern":"try:\n    buffer.extend(other)\nexcept RuntimeError:\n    new_buf = ListBuffer(buffer._coder_impl)\n    new_buf.extend(other)","preventionTips":["Complete all writes before calling partition().","Treat partition() as finalizing the buffer.","Allocate a fresh buffer for post-read data."],"tags":["apache-beam","fn-api-runner","buffer-lifecycle","runtime-error"],"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-20T03:17:13.778Z"}