{"record":{"id":"92c9008a1e80ea7f","repo":"apache/beam","slug":"initializing-threadsafe-requires-a-watermarkestimator","errorCode":null,"errorMessage":"Initializing Threadsafe requires a WatermarkEstimator","messagePattern":"Initializing Threadsafe requires a WatermarkEstimator","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/sdf_utils.py","lineNumber":187,"sourceCode":"\n  def try_claim(self, position):\n    return self._threadsafe_restriction_tracker.try_claim(position)\n\n  def defer_remainder(self, deferred_time=None):\n    self._threadsafe_restriction_tracker.defer_remainder(deferred_time)\n\n  def is_bounded(self):\n    self._threadsafe_restriction_tracker.is_bounded()\n\n\nclass ThreadsafeWatermarkEstimator(object):\n  \"\"\"A threadsafe wrapper which wraps a WatermarkEstimator with locking\n  mechanism to guarantee multi-thread safety.\n  \"\"\"\n  def __init__(self, watermark_estimator: 'WatermarkEstimator') -> None:\n    from apache_beam.io.iobase import WatermarkEstimator\n    if not isinstance(watermark_estimator, WatermarkEstimator):\n      raise ValueError('Initializing Threadsafe requires a WatermarkEstimator')\n    self._watermark_estimator = watermark_estimator\n    self._lock = threading.Lock()\n\n  def __getattr__(self, attr):\n    if hasattr(self._watermark_estimator, attr):\n\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:","sourceCodeStart":169,"sourceCodeEnd":205,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/sdf_utils.py#L169-L205","documentation":"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.","triggerScenarios":"Instantiating Threadsafe(estimator) with None, a raw estimator-like object, or a custom estimator that does not inherit from WatermarkEstimator.","commonSituations":"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.","solutions":["Pass a real WatermarkEstimator instance (e.g. from WatermarkEstimators) to Threadsafe","Make your custom estimator subclass apache_beam.io.iobase.WatermarkEstimator","Call the estimator factory to get an instance rather than passing the factory itself"],"exampleFix":"// before\nestimator = Threadsafe(WatermarkEstimators.Manual)\n// after\nestimator = Threadsafe(WatermarkEstimators.Manual.now())","handlingStrategy":"type-guard","validationCode":"from apache_beam.io.iobase import WatermarkEstimator\nisinstance(x, WatermarkEstimator)  # check before wrapping","typeGuard":"from apache_beam.io.iobase import WatermarkEstimator\ndef is_watermark_estimator(x) -> bool:\n    return isinstance(x, WatermarkEstimator)","tryCatchPattern":"try:\n    est = Threadsafe(x)\nexcept ValueError as e:\n    if 'WatermarkEstimator' in str(e):\n        raise TypeError('must pass a WatermarkEstimator instance') from e\n    raise","preventionTips":["Instantiate the estimator (call the factory) rather than passing the class","Subclass WatermarkEstimator for custom estimators","Type-annotate watermark_estimator() return values in custom SDFs"],"tags":["apache-beam","sdf","type-mismatch","watermark"],"backgroundTag":"invalid-constructor-argument","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"}