{"record":{"id":"74d215d5cd7e97eb","repo":"apache/beam","slug":"unexpected-data-s","errorCode":null,"errorMessage":"Unexpected data: %s","messagePattern":"Unexpected data: (.+?)","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py","lineNumber":518,"sourceCode":"            window.GlobalWindows.windowed_value((b'', merge_context.windows))))\n    to_worker.close()\n\n    process_bundle_req = beam_fn_api_pb2.InstructionRequest(\n        instruction_id=process_bundle_id,\n        process_bundle=beam_fn_api_pb2.ProcessBundleRequest(\n            process_bundle_descriptor_id=self._bundle_processor_id))\n    result_future = worker_handler.control_conn.push(process_bundle_req)\n    for output in worker_handler.data_conn.input_elements(\n        process_bundle_id, [self.FROM_SDK_TRANSFORM],\n        abort_callback=lambda: bool(result_future.is_done() and result_future.\n                                    get().error)):\n      if isinstance(output, beam_fn_api_pb2.Elements.Data):\n        windowed_result = self.windowed_output_coder_impl.decode_nested(\n            output.data)\n        for merge_result, originals in windowed_result.value[1][1]:\n          merge_context.merge(originals, merge_result)\n      else:\n        raise RuntimeError(\"Unexpected data: %s\" % output)\n\n    result = result_future.get()\n    if result.error:\n      raise RuntimeError(result.error)\n    # The result was \"returned\" via the merge callbacks on merge_context above.\n\n  def get_window_coder(self) -> coders.Coder:\n    return self._execution_context_ref().pipeline_context.coders[\n        self._windowing_strategy_proto.window_coder_id]\n\n  def worker_handle(self) -> 'worker_handlers.WorkerHandler':\n    if self._worker_handler is None:\n      worker_handler_manager = self._execution_context_ref(\n      ).worker_handler_manager\n      self._worker_handler = worker_handler_manager.get_worker_handlers(\n          self._windowing_strategy_proto.environment_id, 1)[0]\n      process_bundle_decriptor = self.make_process_bundle_descriptor(\n          self._worker_handler.data_api_service_descriptor(),","sourceCodeStart":500,"sourceCodeEnd":536,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py#L500-L536","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Verify SDK and runner versions match (same Beam release line) so the Elements stream protocol agrees","Check that the windowed output coder for the transform is correctly registered and the payload is Data","Upgrade apache-beam to a version where this merge path handles all element kinds","Report to Beam dev@ if reproducible with a minimal merging-window pipeline"],"exampleFix":"// before\nraise RuntimeError(\"Unexpected data: %s\" % output)\n// after\nif isinstance(output, beam_fn_api_pb2.Elements.Data):\n    ...merge...\nelif isinstance(output, beam_fn_api_pb2.Elements.Timer):\n    pass  # handle/skip non-data elements\nelse:\n    raise RuntimeError(\"Unexpected data: %s\" % output)","handlingStrategy":"try-catch","validationCode":"from apache_beam.runners.portability.fn_api_runner import execution\nisinstance(output, beam_fn_api_pb2.Elements.Data)  # precheck element kind","typeGuard":"def is_data_element(output) -> bool:\n    return isinstance(output, beam_fn_api_pb2.Elements.Data)","tryCatchPattern":"try:\n    ctx.merge(windowing, elements)\nexcept RuntimeError as e:\n    if 'Unexpected data' in str(e):\n        log.error('harness sent non-data element during window merge: %s', e)\n        raise\n    raise","preventionTips":["Keep SDK and runner on the same apache-beam version","Test merging-window pipelines end-to-end before production","Log element types around merge paths in debug builds"],"tags":["python","apache-beam","windows","fn-api-runner"],"backgroundTag":"unexpected-response-shape","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"}