apache/beam · error · ValueError

Initializing Threadsafe requires a WatermarkEstimator

Error message

Initializing Threadsafe requires a WatermarkEstimator

What it means

Threadsafe is a locking wrapper around a WatermarkEstimator for use in splittable DoFns. Its constructor type-checks the argument and raises this ValueError if the object is not an instance of apache_beam.io.iobase.WatermarkEstimator.

Source

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

  def try_claim(self, position):
    return self._threadsafe_restriction_tracker.try_claim(position)

  def defer_remainder(self, deferred_time=None):
    self._threadsafe_restriction_tracker.defer_remainder(deferred_time)

  def is_bounded(self):
    self._threadsafe_restriction_tracker.is_bounded()


class ThreadsafeWatermarkEstimator(object):
  """A threadsafe wrapper which wraps a WatermarkEstimator with locking
  mechanism to guarantee multi-thread safety.
  """
  def __init__(self, watermark_estimator: 'WatermarkEstimator') -> None:
    from apache_beam.io.iobase import WatermarkEstimator
    if not isinstance(watermark_estimator, WatermarkEstimator):
      raise ValueError('Initializing Threadsafe requires a WatermarkEstimator')
    self._watermark_estimator = watermark_estimator
    self._lock = threading.Lock()

  def __getattr__(self, attr):
    if hasattr(self._watermark_estimator, attr):

      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:

View on GitHub (pinned to 12126d8942)

Solutions

1. Pass a real WatermarkEstimator instance (e.g. from WatermarkEstimators) to Threadsafe
2. Make your custom estimator subclass apache_beam.io.iobase.WatermarkEstimator
3. Call the estimator factory to get an instance rather than passing the factory itself

Example fix

// before
estimator = Threadsafe(WatermarkEstimators.Manual)
// after
estimator = Threadsafe(WatermarkEstimators.Manual.now())

from apache_beam.io.iobase import WatermarkEstimator
isinstance(x, WatermarkEstimator)  # check before wrapping

from apache_beam.io.iobase import WatermarkEstimator
def is_watermark_estimator(x) -> bool:
    return isinstance(x, WatermarkEstimator)

try:
    est = Threadsafe(x)
except ValueError as e:
    if 'WatermarkEstimator' in str(e):
        raise TypeError('must pass a WatermarkEstimator instance') from e
    raise

When it happens

Trigger: Instantiating Threadsafe(estimator) with None, a raw estimator-like object, or a custom estimator that does not inherit from WatermarkEstimator.

Common situations: Custom splittable DoFn watermark_estimator() methods returning an object not derived from the iobase.WatermarkEstimator base class; passing the estimator factory instead of an estimator instance.

Related errors

• set_watermark expects a Timestamp as input
• Initialize ThreadsafeRestrictionTracker requiresRestrictionT
• The timestamp of deter_remainder() should be a Duration or a
• Initialize RestrictionTrackerView requires ThreadsafeRestric
• Input of observe_timestamp should be a Timestamp object

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