JonnyIncognito commented on a change in pull request #6210: [AIRFLOW-5567] BaseAsyncOperator URL: https://github.com/apache/airflow/pull/6210#discussion_r339015787
########## File path: airflow/models/base_async_operator.py ########## @@ -0,0 +1,161 @@ +# -*- coding: utf-8 -*- +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +""" +Base Asynchronous Operator for kicking off a long running +operations and polling for completion with reschedule mode. +""" + +from abc import abstractmethod +from typing import Dict, List, Optional, Union + +from airflow.models import SkipMixin, TaskReschedule +from airflow.models.xcom import XCOM_EXTERNAL_RESOURCE_ID_KEY +from airflow.sensors.base_sensor_operator import BaseSensorOperator +from airflow.utils.decorators import apply_defaults + + Review comment: Something else to consider about a potential "multi phase" BaseOperator is that we already have `pre_execute()` + `execute()` + `post_execute()` and the order is governed by `TaskInstance._run_raw_task()` ([link](https://github.com/apache/airflow/blob/master/airflow/models/taskinstance.py#L857)). My understanding is that these hooks are always called even when tasks are rescheduled. I'm not entirely sure of the purpose, the docs simply say "... it’s mostly a hook for people deriving operators" and the only two places I can see in the source-code where they're used is: CloudSqlQueryOperator to ensure the db connection is created and torn down; and for Lineage support to ensure the lists of inlets and outlets are built and saved to Xcom (another kind of state retention!). When considering extra phases, they'd be orthogonal to above and really some form of "execution phase N": the BaseOperator starts to behave a bit like a SubDag where the overall task status / state machine runs its course while what would have been sub-tasks in the SubDag now run as sequential execution phases in the same task. Instead of the sub-tasks being visible in the UI as DAG nodes, these execution phases are internalised into the BaseOperator. Why would we want to do that? The justification is that these execution phases are tightly coupled and should be retried together (and in order) as some kind of atomic unit. If we were to generalise this much, I think we should support being able to run through the execution phases synchronously (taking up a task slot) or asynchronously (using reschedule). Sorry this is getting a bit philosophical but I'm trying to provide some guiding context about why we're doing this! I also see it as a potential future foundation for a mechanism to compose operators via the Airflow API. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services
