{"record":{"id":"535c1c5a648fb40f","repo":"apache/beam","slug":"watermark-must-be-monotonically-increasing-provided","errorCode":null,"errorMessage":"Watermark must be monotonically increasing.Provided watermark %s is less than current watermark %s","messagePattern":"Watermark must be monotonically increasing\\.Provided watermark (.+?) is less than current watermark (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/watermark_estimators.py","lineNumber":132,"sourceCode":"\n  def set_watermark(self, timestamp):\n    # pylint: disable=line-too-long\n\n    \"\"\"Sets a timestamp before or at the timestamps of all future elements\n    produced by the associated DoFn.\n\n    This can be approximate. If records are output that violate this guarantee,\n    they will be considered late, which will affect how they will be processed.\n    See https://beam.apache.org/documentation/programming-guide/#watermarks-and-late-data\n    for more information on late data and how to handle it.\n\n    However, this value should be as late as possible. Downstream windows may\n    not be able to close until this watermark passes their end.\n    \"\"\"\n    if not isinstance(timestamp, Timestamp):\n      raise ValueError('set_watermark expects a Timestamp as input')\n    if self._watermark and self._watermark > timestamp:\n      raise ValueError(\n          'Watermark must be monotonically increasing.'\n          'Provided watermark %s is less than '\n          'current watermark %s',\n          timestamp,\n          self._watermark)\n    self._watermark = timestamp\n\n  @staticmethod\n  def default_provider():\n    \"\"\"Provide a default WatermarkEstimatorProvider for\n    WalltimeWatermarkEstimator.\n    \"\"\"\n    class DefaultManualWatermarkEstimatorProvider(WatermarkEstimatorProvider):\n      def initial_estimator_state(self, element, restriction):\n        return None\n\n      def create_watermark_estimator(self, estimator_state):\n        return ManualWatermarkEstimator(estimator_state)","sourceCodeStart":114,"sourceCodeEnd":150,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/watermark_estimators.py#L114-L150","documentation":"Watermarks must advance monotonically: the runner relies on watermarks never moving backward to trigger windows correctly. ManualWatermarkEstimator.set_watermark raises ValueError when the provided timestamp is earlier than the currently held watermark.","triggerScenarios":"Calling set_watermark with a Timestamp earlier than one set previously — e.g. processing out-of-order events and setting the watermark from each element's event time without taking the max, or recomputing watermarks from scratch on each bundle.","commonSituations":"Out-of-order streaming sources; restarting/replaying data causing older timestamps to arrive after newer ones; computing watermark per-record instead of as a running maximum.","solutions":["Only ever advance: estimator.set_watermark(max(current, new)) — check estimator.current_watermark() before setting","Derive the watermark as a monotone function (e.g. running max of event times) rather than per-element event time","If the earlier timestamp is legitimate, this estimator is the wrong choice — consider a different watermark estimator or let the runner estimate it"],"exampleFix":"// before\nestimator.set_watermark(new_ts)\n// after\nif estimator.current_watermark() is None or new_ts > estimator.current_watermark():\n    estimator.set_watermark(new_ts)","handlingStrategy":"validation","validationCode":"cur = estimator.current_watermark()\nif cur is None or new_ts > cur:\n    estimator.set_watermark(new_ts)","typeGuard":null,"tryCatchPattern":"try:\n    estimator.set_watermark(new_ts)\nexcept ValueError:\n    pass  # stale watermark; safely ignored since it must only move forward","preventionTips":["Always take the running max of event times when computing watermarks","Never reset or recompute watermarks from scratch mid-stream","Remember watermarks are monotone by contract in streaming systems"],"tags":["python","apache-beam","watermark","monotonicity","streaming"],"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"}