apache/beam · error · ValueError

set_watermark expects a Timestamp as input

Error message

set_watermark expects a Timestamp as input

What it means

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.

Source

Thrown at sdks/python/apache_beam/io/watermark_estimators.py:130

  def get_estimator_state(self):
    return self._watermark

  def set_watermark(self, timestamp):
    # pylint: disable=line-too-long

    """Sets a timestamp before or at the timestamps of all future elements
    produced by the associated DoFn.

    This can be approximate. If records are output that violate this guarantee,
    they will be considered late, which will affect how they will be processed.
    See https://beam.apache.org/documentation/programming-guide/#watermarks-and-late-data
    for more information on late data and how to handle it.

    However, this value should be as late as possible. Downstream windows may
    not be able to close until this watermark passes their end.
    """
    if not isinstance(timestamp, Timestamp):
      raise ValueError('set_watermark expects a Timestamp as input')
    if self._watermark and self._watermark > timestamp:
      raise ValueError(
          'Watermark must be monotonically increasing.'
          'Provided watermark %s is less than '
          'current watermark %s',
          timestamp,
          self._watermark)
    self._watermark = timestamp

  @staticmethod
  def default_provider():
    """Provide a default WatermarkEstimatorProvider for
    WalltimeWatermarkEstimator.
    """
    class DefaultManualWatermarkEstimatorProvider(WatermarkEstimatorProvider):
      def initial_estimator_state(self, element, restriction):
        return None

View on GitHub (pinned to 12126d8942)

Solutions

1. Convert to Timestamp first: estimator.set_watermark(Timestamp.now()) or Timestamp(seconds=float_value)
2. For datetime objects use Timestamp.from_rfc3339(dt.isoformat()) or Timestamp(micros=...) For epoch seconds use Timestamp(seconds=int(epoch_secs), nanos=...) or Timestamp(epoch_float)
3. Wrap the call in a type check: isinstance(ts, Timestamp) before calling

Example fix

// before
estimator.set_watermark(time.time())
// after
from apache_beam.utils.timestamp import Timestamp
estimator.set_watermark(Timestamp(time.time()))

from apache_beam.utils.timestamp import Timestamp
if not isinstance(ts, Timestamp):
    ts = Timestamp(ts) if isinstance(ts, (int, float)) else Timestamp.from_rfc3339(ts.isoformat())

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

try:
    estimator.set_watermark(ts)
except ValueError:
    estimator.set_watermark(Timestamp(ts))

When it happens

Trigger: 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).

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

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• repeat(repeats=) value must be an int or a DeferredSeries (e
• Passing a deferred series to round() is not supported, pleas
• str.repeat(repeats=) value must be an int or a DeferredSerie
• Could not find code object with path: {code_object_identifie
• You are using Avro IO with fastavro (default with Beam on Py

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