{"record":{"id":"39c9441826342fce","repo":"apache/beam","slug":"set-watermark-expects-a-timestamp-as-input-estimators","errorCode":null,"errorMessage":"set_watermark expects a Timestamp as input","messagePattern":"set_watermark expects a Timestamp as input","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/watermark_estimators.py","lineNumber":130,"sourceCode":"  def get_estimator_state(self):\n    return self._watermark\n\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","sourceCodeStart":112,"sourceCodeEnd":148,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/watermark_estimators.py#L112-L148","documentation":"ManualWatermarkEstimator.set_watermark requires the input to be an apache_beam.utils.timestamp.Timestamp because watermark arithmetic and runner-API encoding depend on that type. Passing any other type (int seconds, datetime, float) raises ValueError.","triggerScenarios":"Calling estimator.set_watermark(1234567890) with epoch seconds, a datetime.datetime object, a float, or a string inside a DoFn's process (via iobase.DoFn watermarks / RestrictionProvider progress).","commonSituations":"Users deriving watermarks from system time (time.time()) or datetime.now() and passing them directly without conversion; mixing Beam's Timestamp with pandas/datetime objects.","solutions":["Convert to Timestamp first: estimator.set_watermark(Timestamp.now()) or Timestamp(seconds=float_value)","For datetime objects use Timestamp.from_rfc3339(dt.isoformat()) or Timestamp(micros=...)\nFor epoch seconds use Timestamp(seconds=int(epoch_secs), nanos=...) or Timestamp(epoch_float)","Wrap the call in a type check: isinstance(ts, Timestamp) before calling"],"exampleFix":"// before\nestimator.set_watermark(time.time())\n// after\nfrom apache_beam.utils.timestamp import Timestamp\nestimator.set_watermark(Timestamp(time.time()))","handlingStrategy":"type-guard","validationCode":"from apache_beam.utils.timestamp import Timestamp\nif not isinstance(ts, Timestamp):\n    ts = Timestamp(ts) if isinstance(ts, (int, float)) else Timestamp.from_rfc3339(ts.isoformat())","typeGuard":"def is_beam_timestamp(v) -> bool:\n    from apache_beam.utils.timestamp import Timestamp\n    return isinstance(v, Timestamp)","tryCatchPattern":"try:\n    estimator.set_watermark(ts)\nexcept ValueError:\n    estimator.set_watermark(Timestamp(ts))","preventionTips":["Convert time.time()/datetime values to Beam Timestamp at the boundary","Add a unit test asserting the watermark input type","Avoid passing datetime objects into Beam watermark APIs directly"],"tags":["python","apache-beam","type-mismatch","watermark"],"backgroundTag":"type-mismatch","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"}