apache/beam · error · ValueError

Input of observe_timestamp should be a Timestamp object

Error message

Input of observe_timestamp should be a Timestamp object

What it means

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.

Source

Thrown at sdks/python/apache_beam/runners/sdf_utils.py:211

      def method_wrapper(*args, **kw):
        with self._lock:
          return getattr(self._watermark_estimator, attr)(*args, **kw)

      return method_wrapper
    raise AttributeError(attr)

  def get_estimator_state(self):
    with self._lock:
      return self._watermark_estimator.get_estimator_state()

  def current_watermark(self) -> Timestamp:
    with self._lock:
      return self._watermark_estimator.current_watermark()

  def observe_timestamp(self, timestamp: Timestamp) -> None:
    if not isinstance(timestamp, Timestamp):
      raise ValueError(
          'Input of observe_timestamp should be a Timestamp '
          'object')
    with self._lock:
      self._watermark_estimator.observe_timestamp(timestamp)


class NoOpWatermarkEstimatorProvider(WatermarkEstimatorProvider):
  """A WatermarkEstimatorProvider which creates NoOpWatermarkEstimator for the
  framework.
  """
  def initial_estimator_state(self, element, restriction):
    return None

  def create_watermark_estimator(self, estimator_state):
    from apache_beam.io.iobase import WatermarkEstimator

    class _NoOpWatermarkEstimator(WatermarkEstimator):
      """A No-op WatermarkEstimator which is provided for the framework if there

View on GitHub (pinned to 12126d8942)

Solutions

1. Convert to apache_beam.utils.timestamp.Timestamp before observing (e.g. Timestamp(seconds=epoch) or Timestamp.from_rfc3339(...))
2. If using datetime, convert with Timestamp.from_utc_datetime_components or the appropriate helper

Example fix

// before
estimator.observe_timestamp(record.event_time)  # a datetime
// after
from apache_beam.utils.timestamp import Timestamp
estimator.observe_timestamp(Timestamp.from_utc_datetime_components(record.event_time))

from apache_beam.utils.timestamp import Timestamp
isinstance(ts, Timestamp)  # check before observing

from apache_beam.utils.timestamp import Timestamp
def is_beam_timestamp(x) -> bool:
    return isinstance(x, Timestamp)

try:
    estimator.observe_timestamp(v)
except ValueError as e:
    if 'Timestamp object' in str(e):
        estimator.observe_timestamp(Timestamp(int(v)))  # convert epoch
    else:
        raise

When it happens

Trigger: Calling threadsafe_estimator.observe_timestamp(x) with an int, float, datetime.datetime, or None instead of apache_beam.utils.timestamp.Timestamp.

Common situations: 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.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

• The timestamp of deter_remainder() should be a Duration or a
• Unexpected timestamp type: {typeName}
• set_watermark expects a Timestamp as input
• Initialize ThreadsafeRestrictionTracker requiresRestrictionT
• Initialize RestrictionTrackerView requires ThreadsafeRestric

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