apache/beam · error · ValueError

Watermark must be monotonically increasing. Provided %s < cu

Error message

Watermark must be monotonically increasing. Provided %s < current %s

What it means

set_watermark enforces that watermarks only move forward in time. When the provided timestamp is earlier than the currently held watermark, ValueError is raised, because a regressing watermark would corrupt event-time ordering and trigger/pane semantics in the Beam pipeline.

Source

Thrown at sdks/python/apache_beam/io/gcp/bigquery_change_history.py:266

  both values survive SDF re-dispatch.
  """
  def __init__(self, state: tuple[Timestamp, Timestamp]) -> None:
    self._watermark_hold, self._last_end = state

  def observe_timestamp(self, timestamp: Timestamp) -> None:
    pass

  def current_watermark(self) -> Timestamp:
    return self._watermark_hold

  def get_estimator_state(self) -> tuple[Timestamp, Timestamp]:
    return (self._watermark_hold, self._last_end)

  def set_watermark(self, timestamp: Timestamp) -> None:
    if not isinstance(timestamp, Timestamp):
      raise ValueError('set_watermark expects a Timestamp as input')
    if self._watermark_hold and self._watermark_hold > timestamp:
      raise ValueError(
          'Watermark must be monotonically increasing. '
          'Provided %s < current %s' % (timestamp, self._watermark_hold))
    self._watermark_hold = timestamp

  def advance_poll_cursor(self, end: Timestamp) -> None:
    """Record end so the next poll starts from here.

    Only advances forward: if end is earlier than the current cursor
    (e.g. BQ clock regression), the cursor stays put so the next poll
    doesn't re-query an already-covered range.
    """
    self._last_end = max(self._last_end, end)

  def poll_cursor(self) -> Timestamp:
    """Return the start Timestamp for the next poll."""
    return self._last_end

View on GitHub (pinned to 12126d8942)

Solutions

1. Only call set_watermark with timestamps >= the current hold; query get_estimator_state() first and clamp: max(current, new_ts).
2. Ensure query ranges emitted are non-overlapping and monotonically increasing in end time.
3. If recovering from an old checkpoint intentionally, reset the watermark state rather than moving it backwards.

Example fix

// before
state.set_watermark(new_ts)
// after
current, _ = state.get_estimator_state()
if new_ts >= current:
    state.set_watermark(new_ts)

current, _ = state.get_estimator_state()
if current is not None and ts < current:
    ts = current  # clamp instead of regressing
state.set_watermark(ts)

try:
    state.set_watermark(ts)
except ValueError as e:
    log.warning("non-monotonic watermark rejected: %s", e)

When it happens

Trigger: Calling set_watermark(t) where t < self._watermark_hold, e.g. re-processing an older poll range, a query range whose end time overlaps a previously emitted range, or out-of-order watermark updates from _emit_query_ranges / custom estimators.

Common situations: Restarting a stream from an earlier start_time while watermark state persists; overlapping query ranges; passing a stale or cached timestamp; clock issues when polling change history with earlier commit timestamps.

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

• Watermark must be monotonically increasing.Provided watermar
• Please specify a BigQuery table to read from.
• Encountered unsupported parameter(s) in read_gbq: {kwargs.ke
• %s can be set with either schema_update_options or additiona
• The TableRowJsonCoder requires a table schema for encoding o

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