class WatermarkEstimators {

  public static <WatermarkEstimatorStateT>
      WatermarkAndStateObserver<WatermarkEstimatorStateT> threadSafe(
          WatermarkEstimator<WatermarkEstimatorStateT> watermarkEstimator) {
    if (watermarkEstimator instanceof TimestampObservingWatermarkEstimator) {
      return new ThreadSafeTimestampObservingWatermarkEstimator<>(watermarkEstimator);
    } else if (watermarkEstimator instanceof ManualWatermarkEstimator) {
      return new ThreadSafeManualWatermarkEstimator<>(watermarkEstimator);
    }
    return new ThreadSafeWatermarkEstimator<>(watermarkEstimator);
  }

}