{"record":{"id":"629d16915927d2da","repo":"apache/beam","slug":"input-of-observe-timestamp-should-be-a-timestamp-object","errorCode":null,"errorMessage":"Input of observe_timestamp should be a Timestamp object","messagePattern":"Input of observe_timestamp should be a Timestamp object","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/sdf_utils.py","lineNumber":211,"sourceCode":"\n      def method_wrapper(*args, **kw):\n        with self._lock:\n          return getattr(self._watermark_estimator, attr)(*args, **kw)\n\n      return method_wrapper\n    raise AttributeError(attr)\n\n  def get_estimator_state(self):\n    with self._lock:\n      return self._watermark_estimator.get_estimator_state()\n\n  def current_watermark(self) -> Timestamp:\n    with self._lock:\n      return self._watermark_estimator.current_watermark()\n\n  def observe_timestamp(self, timestamp: Timestamp) -> None:\n    if not isinstance(timestamp, Timestamp):\n      raise ValueError(\n          'Input of observe_timestamp should be a Timestamp '\n          'object')\n    with self._lock:\n      self._watermark_estimator.observe_timestamp(timestamp)\n\n\nclass NoOpWatermarkEstimatorProvider(WatermarkEstimatorProvider):\n  \"\"\"A WatermarkEstimatorProvider which creates NoOpWatermarkEstimator for the\n  framework.\n  \"\"\"\n  def initial_estimator_state(self, element, restriction):\n    return None\n\n  def create_watermark_estimator(self, estimator_state):\n    from apache_beam.io.iobase import WatermarkEstimator\n\n    class _NoOpWatermarkEstimator(WatermarkEstimator):\n      \"\"\"A No-op WatermarkEstimator which is provided for the framework if there","sourceCodeStart":193,"sourceCodeEnd":229,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/sdf_utils.py#L193-L229","documentation":"Threadsafe.observe_timestamp() forwards timestamp observations to the wrapped WatermarkEstimator. It validates the input and raises this ValueError if the argument is not an apache_beam Timestamp object.","triggerScenarios":"Calling threadsafe_estimator.observe_timestamp(x) with an int, float, datetime.datetime, or None instead of apache_beam.utils.timestamp.Timestamp.","commonSituations":"Observing event timestamps from records where the field is a unix epoch float or datetime; mixing Beam Timestamp with Python datetime in custom splittable DoFn code.","solutions":["Convert to apache_beam.utils.timestamp.Timestamp before observing (e.g. Timestamp(seconds=epoch) or Timestamp.from_rfc3339(...))","If using datetime, convert with Timestamp.from_utc_datetime_components or the appropriate helper"],"exampleFix":"// before\nestimator.observe_timestamp(record.event_time)  # a datetime\n// after\nfrom apache_beam.utils.timestamp import Timestamp\nestimator.observe_timestamp(Timestamp.from_utc_datetime_components(record.event_time))","handlingStrategy":"type-guard","validationCode":"from apache_beam.utils.timestamp import Timestamp\nisinstance(ts, Timestamp)  # check before observing","typeGuard":"from apache_beam.utils.timestamp import Timestamp\ndef is_beam_timestamp(x) -> bool:\n    return isinstance(x, Timestamp)","tryCatchPattern":"try:\n    estimator.observe_timestamp(v)\nexcept ValueError as e:\n    if 'Timestamp object' in str(e):\n        estimator.observe_timestamp(Timestamp(int(v)))  # convert epoch\n    else:\n        raise","preventionTips":["Convert epoch ints/floats and datetimes to Beam Timestamp at ingestion boundaries","Import Timestamp from apache_beam.utils.timestamp consistently","Unit-test observe_timestamp paths with Timestamp objects only"],"tags":["apache-beam","sdf","type-mismatch","timestamp"],"backgroundTag":"invalid-argument-value","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T21:17:11.552Z"}