apache/beam · error · NotImplementedError

Unsupported merging strategy

Error message

Unsupported merging strategy: %s

What it means

Raised by _make_safe_windowing_strategy when a windowing strategy's merge_status is neither NON_MERGING nor NEEDS_MERGE. The fn-api runner can only wrap merging windows with GenericMergingWindowFn; any other merge status (e.g. UNSPECIFIED or an unknown proto value) is unsupported.

Solutions

1. Ensure the windowing strategy proto has merge_status explicitly set (NON_MERGING or NEEDS_MERGE)
2. Use standard window functions (Fixed/Sliding/SessionWindows) so merge_status is set by the framework
3. Upgrade apache-beam if your pipeline uses a MergeStatus value added after your runner version
4. Inspect the WindowingStrategy proto (e.g. via pipeline.to_runner_api()) to see the offending merge_status

Example fix

// before
proto = beam_runner_api_pb2.WindowingStrategy()  # merge_status unset
// after
proto = beam_runner_api_pb2.WindowingStrategy(merge_status=beam_runner_api_pb2.MergeStatus.NEEDS_MERGE)

ws = pipeline_proto.components.windowing_strategies[wid]
assert ws.merge_status in (MergeStatus.NON_MERGING, MergeStatus.NEEDS_MERGE), ws.merge_status

def has_supported_merge_status(ws) -> bool:
    return ws.merge_status in (beam_runner_api_pb2.MergeStatus.NON_MERGING,
                               beam_runner_api_pb2.MergeStatus.NEEDS_MERGE)

try:
    stage = translations.create_and_optimize_stages(proto, ...)
except NotImplementedError as e:
    if 'Unsupported merging strategy' in str(e):
        fix_windowing_strategy(pipeline)
    raise

When it happens

Trigger: Building a FnApiRunner stage whose windowing strategy proto has a merge_status value outside the two supported ones — typically an uninitialized strategy proto or a newer proto enum value the runner doesn't know.

Common situations: Custom windowing with an incompletely populated WindowingStrategy proto; pipelines built programmatically skipping windowing strategy defaults; proto version drift adding new MergeStatus values.

Related errors

• No existing_windows in this context.
• Unexpected data
• 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/29aaebbb7f6ed7bf. Report an issue: GitHub.

Appendix: source

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

    if windowing_strategy_proto.window_fn.urn in SAFE_WINDOW_FNS:
      return id
    else:
      safe_id = id + '_safe'
      while safe_id in self.pipeline_components.windowing_strategies:
        safe_id += '_'
      safe_proto = copy.copy(windowing_strategy_proto)
      if (windowing_strategy_proto.merge_status ==
          beam_runner_api_pb2.MergeStatus.NON_MERGING):
        safe_proto.window_fn.urn = GenericNonMergingWindowFn.URN
        safe_proto.window_fn.payload = (
            windowing_strategy_proto.window_coder_id.encode('utf-8'))
      elif (windowing_strategy_proto.merge_status ==
            beam_runner_api_pb2.MergeStatus.NEEDS_MERGE):
        window_fn = GenericMergingWindowFn(self, windowing_strategy_proto)
        safe_proto.window_fn.urn = GenericMergingWindowFn.URN
        safe_proto.window_fn.payload = window_fn.payload()
      else:
        raise NotImplementedError(
            'Unsupported merging strategy: %s' %
            windowing_strategy_proto.merge_status)
      self.pipeline_context.windowing_strategies.put_proto(safe_id, safe_proto)
      return safe_id

  @property
  def state_servicer(self) -> 'worker_handlers.StateServicer':
    # TODO(BEAM-9625): Ensure FnApiRunnerExecutionContext owns StateServicer
    return self.worker_handler_manager.state_servicer

  def next_uid(self) -> str:
    self._last_uid += 1
    return str(self._last_uid)

  def _iterable_state_write(
      self, values: Iterable, element_coder_impl: CoderImpl) -> bytes:
    token = unique_name(None, 'iter').encode('ascii')
    out = create_OutputStream()

View on GitHub (pinned to 12126d8942)