apache/beam · error · RuntimeError

Unexpected data

Error message

Unexpected data: %s

What it means

Raised in FnApiRunnerExecutionContext.merge when the harness returns a result element that is not an Elements.Data message while applying windowed merge results. The merge path only knows how to decode Data payloads feeding window merge callbacks; anything else (timers, control messages, or empty/unexpected responses) is a protocol violation.

Solutions

1. Verify SDK and runner versions match (same Beam release line) so the Elements stream protocol agrees
2. Check that the windowed output coder for the transform is correctly registered and the payload is Data
3. Upgrade apache-beam to a version where this merge path handles all element kinds
4. Report to Beam dev@ if reproducible with a minimal merging-window pipeline

Example fix

// before
raise RuntimeError("Unexpected data: %s" % output)
// after
if isinstance(output, beam_fn_api_pb2.Elements.Data):
    ...merge...
elif isinstance(output, beam_fn_api_pb2.Elements.Timer):
    pass  # handle/skip non-data elements
else:
    raise RuntimeError("Unexpected data: %s" % output)

from apache_beam.runners.portability.fn_api_runner import execution
isinstance(output, beam_fn_api_pb2.Elements.Data)  # precheck element kind

def is_data_element(output) -> bool:
    return isinstance(output, beam_fn_api_pb2.Elements.Data)

try:
    ctx.merge(windowing, elements)
except RuntimeError as e:
    if 'Unexpected data' in str(e):
        log.error('harness sent non-data element during window merge: %s', e)
        raise
    raise

When it happens

Trigger: Calling merge() on a merging-window context when the bundle result future contains non-Data Elements in the output stream being inspected, e.g. the SDK returned control/timer elements where windowed output data was expected.

Common situations: Runner/SDK version mismatch where the harness emits a different Elements stream shape; corrupted or reordered bundle responses during windowed (e.g. sliding/session) merges.

Related errors

• No existing_windows in this context.
• Unsupported merging strategy
• A BigQuery table or a query must be specified
• A cluster_identifier should be Optional[Union[str…
• A context manager constructor (not a fully constructed…

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

Appendix: source

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

            window.GlobalWindows.windowed_value((b'', merge_context.windows))))
    to_worker.close()

    process_bundle_req = beam_fn_api_pb2.InstructionRequest(
        instruction_id=process_bundle_id,
        process_bundle=beam_fn_api_pb2.ProcessBundleRequest(
            process_bundle_descriptor_id=self._bundle_processor_id))
    result_future = worker_handler.control_conn.push(process_bundle_req)
    for output in worker_handler.data_conn.input_elements(
        process_bundle_id, [self.FROM_SDK_TRANSFORM],
        abort_callback=lambda: bool(result_future.is_done() and result_future.
                                    get().error)):
      if isinstance(output, beam_fn_api_pb2.Elements.Data):
        windowed_result = self.windowed_output_coder_impl.decode_nested(
            output.data)
        for merge_result, originals in windowed_result.value[1][1]:
          merge_context.merge(originals, merge_result)
      else:
        raise RuntimeError("Unexpected data: %s" % output)

    result = result_future.get()
    if result.error:
      raise RuntimeError(result.error)
    # The result was "returned" via the merge callbacks on merge_context above.

  def get_window_coder(self) -> coders.Coder:
    return self._execution_context_ref().pipeline_context.coders[
        self._windowing_strategy_proto.window_coder_id]

  def worker_handle(self) -> 'worker_handlers.WorkerHandler':
    if self._worker_handler is None:
      worker_handler_manager = self._execution_context_ref(
      ).worker_handler_manager
      self._worker_handler = worker_handler_manager.get_worker_handlers(
          self._windowing_strategy_proto.environment_id, 1)[0]
      process_bundle_decriptor = self.make_process_bundle_descriptor(
          self._worker_handler.data_api_service_descriptor(),

View on GitHub (pinned to 12126d8942)