apache/beam · error · RuntimeError

Grouping table append after read.

Error message

Grouping table append after read.

What it means

GroupingBuffer.append() raises RuntimeError if grouped output was already produced (self._grouped_output set). Once results have been read out of the grouping table, further raw inserts would bypass the grouping pipeline and corrupt consistency, so the append is rejected.

Solutions

1. Ensure all input data is appended before reading grouped output.
2. Create a new GroupingBuffer for data arriving after the read.
3. Restructure the runner loop to fully drain input before materializing grouped results.

Example fix

# before
results = list(gbuf)
gbuf.append(data)  # RuntimeError
# after
gbuf.append(data)
results = list(gbuf)

if gbuf._grouped_output is None:
    gbuf.append(data)

def accepts_input(gbuf):
    return not gbuf._grouped_output

try:
    gbuf.append(data)
except RuntimeError:
    new_gbuf = GroupingBuffer(...)
    new_gbuf.append(data)

When it happens

Trigger: Calling grouping_buffer.append(elements_data) after the grouped results have been read (e.g. after __iter__/output extraction set _grouped_output), such as late elements arriving for a finished grouping stage.

Common situations: Streaming-style feeds pushing data into a GroupingBuffer after it was flushed; runner code that reads grouped output and then processes remaining input bytes against the same buffer.

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

• ListBuffer 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/1635d7f154b4b0c3. Report an issue: GitHub.

Appendix: source

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

      windowing: core.Windowing) -> None:
    self._key_coder = pre_grouped_coder.key_coder()
    self._pre_grouped_coder = pre_grouped_coder
    self._post_grouped_coder = post_grouped_coder
    self._table: collections.defaultdict[bytes,
                                         list[Any]] = collections.defaultdict(
                                             list)
    self._windowing = windowing
    self._grouped_output: Optional[list[list[bytes]]] = None

  def copy(self) -> 'GroupingBuffer':
    # This is a silly temporary optimization. This class must be removed once
    # full support for streaming is added (i.e. once we use trigger_manager for
    # data grouping instead of GroupingBuffer).
    return self

  def append(self, elements_data: bytes) -> None:
    if self._grouped_output:
      raise RuntimeError('Grouping table append after read.')
    input_stream = create_InputStream(elements_data)
    coder_impl = self._pre_grouped_coder.get_impl()
    key_coder_impl = self._key_coder.get_impl()
    # TODO(robertwb): We could optimize this even more by using a
    # window-dropping coder for the data plane.
    is_trivial_windowing = self._windowing.is_default()
    while input_stream.size() > 0:
      windowed_key_value = coder_impl.decode_from_stream(input_stream, True)
      key, value = windowed_key_value.value
      self._table[key_coder_impl.encode(key)].append(
          value if is_trivial_windowing else windowed_key_value.
          with_value(value))

  def extend(self, input_buffer: Buffer) -> None:
    if isinstance(input_buffer, ListBuffer):
      # TODO(pabloem): GroupingBuffer will be removed once shuffling is done
      #  via state. Remove this workaround along with that.
      return

View on GitHub (pinned to 12126d8942)