{"record":{"id":"69756c38c22bb0a3","repo":"apache/beam","slug":"watermark-must-be-monotonically-increasing-provided-s","errorCode":null,"errorMessage":"Watermark must be monotonically increasing. Provided %s < current %s","messagePattern":"Watermark must be monotonically increasing\\. Provided (.+?) < current (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/gcp/bigquery_change_history.py","lineNumber":266,"sourceCode":"  both values survive SDF re-dispatch.\n  \"\"\"\n  def __init__(self, state: tuple[Timestamp, Timestamp]) -> None:\n    self._watermark_hold, self._last_end = state\n\n  def observe_timestamp(self, timestamp: Timestamp) -> None:\n    pass\n\n  def current_watermark(self) -> Timestamp:\n    return self._watermark_hold\n\n  def get_estimator_state(self) -> tuple[Timestamp, Timestamp]:\n    return (self._watermark_hold, self._last_end)\n\n  def set_watermark(self, timestamp: Timestamp) -> None:\n    if not isinstance(timestamp, Timestamp):\n      raise ValueError('set_watermark expects a Timestamp as input')\n    if self._watermark_hold and self._watermark_hold > timestamp:\n      raise ValueError(\n          'Watermark must be monotonically increasing. '\n          'Provided %s < current %s' % (timestamp, self._watermark_hold))\n    self._watermark_hold = timestamp\n\n  def advance_poll_cursor(self, end: Timestamp) -> None:\n    \"\"\"Record end so the next poll starts from here.\n\n    Only advances forward: if end is earlier than the current cursor\n    (e.g. BQ clock regression), the cursor stays put so the next poll\n    doesn't re-query an already-covered range.\n    \"\"\"\n    self._last_end = max(self._last_end, end)\n\n  def poll_cursor(self) -> Timestamp:\n    \"\"\"Return the start Timestamp for the next poll.\"\"\"\n    return self._last_end\n\n","sourceCodeStart":248,"sourceCodeEnd":284,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/gcp/bigquery_change_history.py#L248-L284","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Only call set_watermark with timestamps >= the current hold; query get_estimator_state() first and clamp: max(current, new_ts).","Ensure query ranges emitted are non-overlapping and monotonically increasing in end time.","If recovering from an old checkpoint intentionally, reset the watermark state rather than moving it backwards."],"exampleFix":"// before\nstate.set_watermark(new_ts)\n// after\ncurrent, _ = state.get_estimator_state()\nif new_ts >= current:\n    state.set_watermark(new_ts)","handlingStrategy":"validation","validationCode":"current, _ = state.get_estimator_state()\nif current is not None and ts < current:\n    ts = current  # clamp instead of regressing\nstate.set_watermark(ts)","typeGuard":null,"tryCatchPattern":"try:\n    state.set_watermark(ts)\nexcept ValueError as e:\n    log.warning(\"non-monotonic watermark rejected: %s\", e)","preventionTips":["Read get_estimator_state() before advancing and clamp to max(current, new)","Ensure query ranges are non-overlapping and strictly increasing","Never feed stale or cached timestamps into set_watermark"],"tags":["python","apache-beam","bigquery","watermark","monotonicity"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}